"""Focused tests for the Hermes shared-metrics durable store."""

from __future__ import annotations

import json
import multiprocessing as mp
import os
import sqlite3
import stat
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from copy import deepcopy
from datetime import datetime, timezone
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest
from agent import relay_runtime
from hermes_cli.observability import shared_metrics as shared_metrics_module
from hermes_cli.observability.shared_metrics import SharedMetricsStore
from hermes_cli.observability.shared_metrics_contract import (
    COUNT_BUCKETS,
    DURATION_BUCKETS,
    EXECUTION_SURFACES,
    LEGACY_MODEL_CALL_METRIC,
    MODEL_CALL_PROFILE_MODEL,
    MODEL_IDENTIFIER_MAX_LENGTH,
    MODEL_ROUTE_METRIC,
    PROVIDER_IDENTIFIER_MAX_LENGTH,
    SCHEMA_KEY,
    SCHEMA_VERSION,
    TASK_END_REASONS,
    TASK_ENTRYPOINTS,
    TASK_OUTCOMES,
    TASK_TERMINATIONS,
    count_bucket,
    duration_bucket,
    execution_surface,
    model_call_dimensions,
    model_call_fields,
    task_counter,
    task_start_fields,
    task_terminal_fields,
    task_terminal_state,
)


SCHEMA_PATH = (
    Path(__file__).resolve().parents[2]
    / "hermes_cli"
    / "observability"
    / "schemas"
    / "hermes.shared_metrics.v2.schema.json"
)
LEGACY_SCHEMA_PATH = SCHEMA_PATH.with_name("hermes.shared_metrics.v1.schema.json")


def _schema_validator(path: Path = SCHEMA_PATH):
    jsonschema = pytest.importorskip("jsonschema")
    schema = json.loads(path.read_text(encoding="utf-8"))
    jsonschema.Draft202012Validator.check_schema(schema)
    return jsonschema.Draft202012Validator(
        schema,
        format_checker=jsonschema.FormatChecker(),
    )


def _package_dimension_schema() -> dict[str, object]:
    schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
    return schema["$defs"]["model_route_counter"]["properties"]["dimensions"]


def _task_dimension_schema(kind: str) -> dict[str, object]:
    schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
    return schema["$defs"][kind]["properties"]["dimensions"]


def _dimensions() -> dict[str, str]:
    return {
        "model": "anthropic/claude-sonnet-4.6",
        "provider": "openrouter",
    }


def _legacy_dimensions() -> dict[str, str]:
    return {
        "call_role": "primary",
        "locality": "remote",
        "model_family": "claude",
        "outcome": "success",
        "provider_family": "direct",
    }


def _record_model_calls_in_process(
    database_path: str,
    outbox_directory: str,
    count: int,
    start_barrier: Any | None = None,
) -> None:
    if start_barrier is not None:
        start_barrier.wait()
    store = SharedMetricsStore(Path(database_path), Path(outbox_directory))
    for _ in range(count):
        store.record_model_call(_dimensions(), "test-version")


def test_model_call_counter_survives_restart_and_exports_only_new_deltas(tmp_path):
    database_path = tmp_path / "metrics.sqlite3"
    outbox_directory = tmp_path / "outbox"
    store = SharedMetricsStore(database_path, outbox_directory)
    store.record_model_call(_dimensions(), "test-version")
    store.record_model_call(_dimensions(), "test-version")

    first_paths = store.create_and_export_package()

    assert len(first_paths) == 1
    first_package = json.loads(first_paths[0].read_text(encoding="utf-8"))
    _schema_validator().validate(first_package)
    uuid.UUID(first_package["package_id"])
    uuid.UUID(first_package["install_id"])
    assert first_package["schema_version"] == "hermes.shared_metrics.v2"
    assert first_package["resource"] == {"hermes_version": "test-version"}
    assert first_package["metrics"] == [
        {
            "name": MODEL_ROUTE_METRIC,
            "type": "counter",
            "dimensions": _dimensions(),
            "value": 2,
        }
    ]

    restarted = SharedMetricsStore(database_path, outbox_directory)
    assert restarted.counter_snapshot()[0]["value"] == 2
    assert restarted.counter_snapshot()[0]["packaged_value"] == 2
    assert restarted.create_and_export_package() == []
    assert len(list(outbox_directory.glob("*.json"))) == 1

    restarted.record_model_call(_dimensions(), "test-version")
    second_paths = restarted.create_and_export_package()

    assert len(second_paths) == 1
    second_package = json.loads(second_paths[0].read_text(encoding="utf-8"))
    assert second_package["package_id"] != first_package["package_id"]
    assert second_package["install_id"] == first_package["install_id"]
    assert second_package["metrics"][0]["value"] == 1
    assert restarted.counter_snapshot()[0]["value"] == 3
    assert restarted.counter_snapshot()[0]["packaged_value"] == 3


def test_v2_package_preserves_pending_v1_model_counters(tmp_path):
    database_path = tmp_path / "metrics.sqlite3"
    outbox_directory = tmp_path / "outbox"
    store = SharedMetricsStore(database_path, outbox_directory)
    period_start = shared_metrics_module._utc_now().date().isoformat()
    legacy_dimensions_json = json.dumps(
        _legacy_dimensions(),
        sort_keys=True,
        separators=(",", ":"),
    )
    with sqlite3.connect(database_path) as connection:
        connection.execute(
            """
            INSERT INTO counter_aggregates(
                period_start,
                metric_name,
                hermes_version,
                dimensions_json,
                value,
                packaged_value
            ) VALUES (?, ?, ?, ?, 3, 0)
            """,
            (
                period_start,
                LEGACY_MODEL_CALL_METRIC,
                "test-version",
                legacy_dimensions_json,
            ),
        )
    store.record_model_call(_dimensions(), "test-version")

    [package_path] = store.create_and_export_package()
    package = json.loads(package_path.read_text(encoding="utf-8"))
    _schema_validator().validate(package)

    assert package["schema_version"] == "hermes.shared_metrics.v2"
    assert package["metrics"] == [
        {
            "name": LEGACY_MODEL_CALL_METRIC,
            "type": "counter",
            "dimensions": _legacy_dimensions(),
            "value": 3,
        },
        {
            "name": MODEL_ROUTE_METRIC,
            "type": "counter",
            "dimensions": _dimensions(),
            "value": 1,
        },
    ]
    assert all(
        row["value"] == row["packaged_value"] for row in store.counter_snapshot()
    )


def test_v1_outbox_package_exports_unchanged_after_upgrade(tmp_path):
    database_path = tmp_path / "metrics.sqlite3"
    outbox_directory = tmp_path / "outbox"
    store = SharedMetricsStore(database_path, outbox_directory)
    package_id = str(uuid.uuid4())
    payload = {
        "schema_version": "hermes.shared_metrics.v1",
        "package_id": package_id,
        "install_id": str(uuid.uuid4()),
        "period_start": "2026-07-28T00:00:00Z",
        "period_end": "2026-07-29T00:00:00Z",
        "generated_at": "2026-07-29T01:00:00Z",
        "resource": {"hermes_version": "legacy-version"},
        "metrics": [
            {
                "name": LEGACY_MODEL_CALL_METRIC,
                "type": "counter",
                "dimensions": _legacy_dimensions(),
                "value": 2,
            }
        ],
    }
    payload_json = json.dumps(payload, sort_keys=True, separators=(",", ":"))
    _schema_validator(LEGACY_SCHEMA_PATH).validate(payload)
    with sqlite3.connect(database_path) as connection:
        connection.execute(
            """
            INSERT INTO package_outbox(
                package_id,
                period_start,
                period_end,
                payload_json,
                created_at
            ) VALUES (?, ?, ?, ?, ?)
            """,
            (
                package_id,
                payload["period_start"],
                payload["period_end"],
                payload_json,
                payload["generated_at"],
            ),
        )

    assert store.create_and_export_package() == [
        outbox_directory / f"{package_id}.json"
    ]
    exported = json.loads(
        (outbox_directory / f"{package_id}.json").read_text(encoding="utf-8")
    )
    assert exported == payload
    with sqlite3.connect(database_path) as connection:
        [persisted_payload_json] = connection.execute(
            "SELECT payload_json FROM package_outbox WHERE package_id = ?",
            (package_id,),
        ).fetchone()
    assert persisted_payload_json == payload_json


def test_due_export_runs_once_per_utc_day_and_catches_up_pending_deltas(
    tmp_path, monkeypatch
):
    current_time = datetime(2026, 7, 28, 9, tzinfo=timezone.utc)
    monkeypatch.setattr(shared_metrics_module, "_utc_now", lambda: current_time)
    store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")

    store.record_model_call(_dimensions(), "test-version")
    assert len(store.create_and_export_package_if_due()) == 1

    current_time = datetime(2026, 7, 28, 18, tzinfo=timezone.utc)
    store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")
    store.record_model_call(_dimensions(), "test-version")
    assert store.create_and_export_package_if_due() == []
    assert len(list((tmp_path / "outbox").glob("*.json"))) == 1
    assert store.counter_snapshot()[0] == {
        "period_start": "2026-07-28",
        "metric_name": MODEL_ROUTE_METRIC,
        "hermes_version": "test-version",
        "dimensions": _dimensions(),
        "value": 2,
        "packaged_value": 1,
    }

    current_time = datetime(2026, 7, 29, 9, tzinfo=timezone.utc)
    store.record_model_call(_dimensions(), "test-version")
    assert len(store.create_and_export_package_if_due()) == 2
    assert len(list((tmp_path / "outbox").glob("*.json"))) == 3
    assert all(
        row["value"] == row["packaged_value"] for row in store.counter_snapshot()
    )

    store.record_model_call(_dimensions(), "test-version")
    assert store.create_and_export_package_if_due() == []
    assert len(list((tmp_path / "outbox").glob("*.json"))) == 3


def test_package_schema_matches_the_model_call_contract():
    schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8"))
    properties = _package_dimension_schema()["properties"]

    assert schema["properties"]["schema_version"]["const"] == "hermes.shared_metrics.v2"
    assert set(properties) == {"model", "provider"}
    assert properties["model"]["maxLength"] == MODEL_IDENTIFIER_MAX_LENGTH
    assert properties["provider"]["maxLength"] == PROVIDER_IDENTIFIER_MAX_LENGTH
    assert "enum" not in properties["model"]
    assert "enum" not in properties["provider"]


def test_v1_package_schema_retains_the_legacy_model_contract():
    schema = json.loads(LEGACY_SCHEMA_PATH.read_text(encoding="utf-8"))
    model_counter = schema["$defs"]["model_call_counter"]

    assert schema["properties"]["schema_version"]["const"] == "hermes.shared_metrics.v1"
    assert model_counter["properties"]["name"]["const"] == LEGACY_MODEL_CALL_METRIC
    assert set(model_counter["properties"]["dimensions"]["properties"]) == {
        "call_role",
        "locality",
        "model_family",
        "outcome",
        "provider_family",
    }


def test_model_call_fields_report_terminal_model_and_provider_without_a_catalog():
    assert model_call_fields({
        "model": "fallback/model",
        "response_model": "NVIDIA/Nemotron-3-Ultra",
        "provider": "OpenRouter",
        "base_url": "https://private-endpoint.example/v1",
    }) == {
        "model": "nvidia/nemotron-3-ultra",
        "provider": "openrouter",
    }
    assert model_call_fields({
        "model": "ZAI/GLM-5.2",
        "provider": "Brev",
    }) == {
        "model": "zai/glm-5.2",
        "provider": "brev",
    }


def test_auxiliary_logical_scope_projects_one_normalized_terminal_route():
    event = SimpleNamespace(
        kind="scope",
        category="function",
        name=relay_runtime.LOGICAL_LLM_SCOPE,
        scope_category="end",
        category_profile=None,
        data={
            "model": "Accepted/Model",
            "outcome": "success",
            "provider": "OpenRouter",
        },
        metadata={
            relay_runtime.RUNTIME_SCHEMA_KEY: relay_runtime.RUNTIME_SCHEMA_VERSION,
            relay_runtime.RUNTIME_INSTANCE_KEY: "runtime-1",
            "hermes.call_role": "auxiliary:compression",
        },
    )

    assert model_call_dimensions(event) == {
        "model": "accepted/model",
        "provider": "openrouter",
    }

    event.data.update({
        "model": "configured/model",
        "response_model": "malformed response model",
    })
    assert model_call_dimensions(event) == {
        "model": "configured/model",
        "provider": "openrouter",
    }

    event.metadata["hermes.call_role"] = "primary"
    assert model_call_dimensions(event) is None


@pytest.mark.parametrize(
    "response_model",
    [
        "contains a space",
        "x" * (MODEL_IDENTIFIER_MAX_LENGTH + 1),
    ],
)
def test_model_call_fields_fall_back_when_response_model_is_invalid(response_model):
    assert model_call_fields({
        "model": "nvidia/nemotron-3-ultra",
        "response_model": response_model,
        "provider": "openrouter",
    }) == {
        "model": "nvidia/nemotron-3-ultra",
        "provider": "openrouter",
    }


@pytest.mark.parametrize(
    ("field", "value"),
    [
        ("model", ""),
        ("model", "contains a space"),
        ("model", "contains\ncontrol"),
        ("model", "_" + "private"),
        ("model", "x" * (MODEL_IDENTIFIER_MAX_LENGTH + 1)),
        ("model", object()),
        ("provider", ""),
        ("provider", "private provider"),
        ("provider", "x" * (PROVIDER_IDENTIFIER_MAX_LENGTH + 1)),
        ("provider", object()),
    ],
)
def test_model_call_fields_collapse_malformed_identifiers(field, value):
    event = {"model": "nvidia/nemotron-3-ultra", "provider": "openrouter"}
    event[field] = value

    assert model_call_fields(event)[field] == "unknown"


def test_store_does_not_record_the_retired_model_metric(tmp_path):
    store = SharedMetricsStore(tmp_path / "metrics.sqlite3", tmp_path / "outbox")

    with pytest.raises(ValueError, match="Unsupported shared metric"):
        store.record_counter(
            LEGACY_MODEL_CALL_METRIC,
            _legacy_dimensions(),
            "test-version",
        )

    assert store.counter_snapshot() == []
































def test_retention_prunes_only_expired_exported_history(tmp_path):
    database_path = tmp_path / "metrics.sqlite3"
    outbox_directory = tmp_path / "outbox"
    store = SharedMetricsStore(database_path, outbox_directory)

    store.record_model_call(_dimensions(), "expired-version")
    [expired_path] = store.create_and_export_package()
    store.record_model_call(_dimensions(), "current-version")
    [current_path] = store.create_and_export_package()
    store.record_model_call(_dimensions(), "pending-version")
    pending_package = store._create_package()
    assert pending_package is not None

    with sqlite3.connect(database_path) as connection:
        connection.execute(
            """
            UPDATE counter_aggregates
            SET period_start = '2026-05-01'
            WHERE hermes_version = 'expired-version'
            """
        )
        connection.execute(
            """
            UPDATE package_outbox
            SET period_start = '2026-05-01T00:00:00Z',
                period_end = '2026-05-02T00:00:00Z',
                exported_at = '2026-05-02T00:00:00Z'
            WHERE package_id = ?
            """,
            (expired_path.stem,),
        )

    store._prune_expired_history(
        now=datetime(2026, 7, 23, tzinfo=timezone.utc)
    )

    assert not expired_path.exists()
    assert current_path.exists()
    assert not (outbox_directory / f"{pending_package['package_id']}.json").exists()
    with sqlite3.connect(database_path) as connection:
        outbox_rows = connection.execute(
            "SELECT package_id, exported_at FROM package_outbox ORDER BY package_id"
        ).fetchall()
        aggregate_versions = {
            row[0]
            for row in connection.execute(
                "SELECT hermes_version FROM counter_aggregates"
            ).fetchall()
        }
    assert {row[0] for row in outbox_rows} == {
        current_path.stem,
        pending_package["package_id"],
    }
    assert next(
        row[1] for row in outbox_rows if row[0] == pending_package["package_id"]
    ) is None
    assert aggregate_versions == {"current-version", "pending-version"}








def test_concurrent_package_builders_commit_one_delta(tmp_path):
    database_path = tmp_path / "metrics.sqlite3"
    outbox_directory = tmp_path / "outbox"
    store = SharedMetricsStore(database_path, outbox_directory)
    store.record_model_call(_dimensions(), "test-version")
    ready = threading.Barrier(2)

    def export() -> list[Path]:
        worker_store = SharedMetricsStore(database_path, outbox_directory)
        ready.wait(timeout=5)
        return worker_store.create_and_export_package()

    with ThreadPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit(export) for _ in range(2)]
        for future in futures:
            future.result()

    with sqlite3.connect(database_path) as connection:
        [outbox_count] = connection.execute(
            "SELECT COUNT(*) FROM package_outbox"
        ).fetchone()
    [package_path] = list(outbox_directory.glob("*.json"))
    package = json.loads(package_path.read_text(encoding="utf-8"))

    assert outbox_count == 1
    assert package["metrics"][0]["value"] == 1
    assert store.counter_snapshot()[0]["packaged_value"] == 1






def test_cross_process_model_call_updates_are_transactional(tmp_path):
    database_path = tmp_path / "metrics.sqlite3"
    outbox_directory = tmp_path / "outbox"
    context = mp.get_context("spawn")
    start_barrier = context.Barrier(2)
    processes = [
        context.Process(
            target=_record_model_calls_in_process,
            args=(str(database_path), str(outbox_directory), 10, start_barrier),
        )
        for _ in range(2)
    ]

    for process in processes:
        process.start()
    for process in processes:
        process.join(timeout=15)
        assert not process.is_alive()
        assert process.exitcode == 0

    restarted = SharedMetricsStore(database_path, outbox_directory)
    assert restarted.counter_snapshot()[0]["value"] == 20




@pytest.mark.skipif(os.name == "nt", reason="POSIX permission modes are unavailable")
def test_store_and_export_are_owner_only(tmp_path):
    database_path = tmp_path / "private-store" / "metrics.sqlite3"
    outbox_directory = tmp_path / "private-outbox"
    store = SharedMetricsStore(database_path, outbox_directory)
    store.record_model_call(_dimensions(), "test-version")
    [package_path] = store.create_and_export_package()

    assert stat.S_IMODE(database_path.parent.stat().st_mode) == 0o700
    assert stat.S_IMODE(outbox_directory.stat().st_mode) == 0o700
    assert stat.S_IMODE(database_path.stat().st_mode) == 0o600
    assert stat.S_IMODE(package_path.stat().st_mode) == 0o600