"""Step definitions for ContextTierService thread-safety tests (#7547).

Verifies that all public methods of ContextTierService are safe for
concurrent access from multiple threads, preventing:
- RuntimeError: dictionary changed size during iteration
- Data corruption in hot/warm/cold tier dicts
- Lost updates or phantom reads under concurrent plan execution
"""

from __future__ import annotations

import threading
from datetime import UTC, datetime, timedelta
from typing import Any

from behave import given, then, when

from cleveragents.application.services.context_tiers import ContextTierService
from cleveragents.domain.models.acms.tiers import ContextTier, TieredFragment

__all__: list[str] = []

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

_THREAD_COUNT_DEFAULT = 10


def _make_fragment(
    fid: str,
    tier: ContextTier = ContextTier.HOT,
    token_count: int = 10,
    stale: bool = False,
) -> TieredFragment:
    """Create a TieredFragment for testing."""
    frag = TieredFragment(
        fragment_id=fid,
        content=f"content-{fid}",
        tier=tier,
        token_count=token_count,
    )
    if stale:
        frag.last_accessed = datetime.now(tz=UTC) - timedelta(hours=48)
    return frag


# ---------------------------------------------------------------------------
# Background
# ---------------------------------------------------------------------------


@given("a thread-safe context tier service")
def step_given_thread_safe_service(context: Any) -> None:
    context.tier_service = ContextTierService()
    context.errors = []


# ---------------------------------------------------------------------------
# Lock attribute checks
# ---------------------------------------------------------------------------


@then("the tier service should have a _lock attribute")
def step_then_has_lock(context: Any) -> None:
    assert hasattr(context.tier_service, "_lock"), (
        "ContextTierService must have a _lock attribute"
    )


@then("the _lock should be a threading.RLock")
def step_then_lock_is_rlock(context: Any) -> None:
    # threading.RLock() returns an instance of _RLock (internal type).
    # We verify it has the acquire/release interface of a reentrant lock.
    lock = context.tier_service._lock
    assert hasattr(lock, "acquire"), "_lock must have acquire()"
    assert hasattr(lock, "release"), "_lock must have release()"
    # Verify reentrancy: acquiring twice from the same thread must not deadlock.
    lock.acquire()
    lock.acquire()
    lock.release()
    lock.release()


# ---------------------------------------------------------------------------
# Concurrent store
# ---------------------------------------------------------------------------


@when("{n:d} threads concurrently store {k:d} fragments each into the tier service")
def step_when_concurrent_store(context: Any, n: int, k: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()

    def worker(thread_idx: int) -> None:
        for i in range(k):
            fid = f"ts-store-t{thread_idx}-f{i}"
            frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5)
            try:
                context.tier_service.store(frag)
            except Exception as exc:
                with lock:
                    errors.append(exc)

    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no RuntimeError should have been raised during concurrent store")
def step_then_no_runtime_error_store(context: Any) -> None:
    runtime_errors = [e for e in context.errors if isinstance(e, RuntimeError)]
    assert not runtime_errors, (
        f"RuntimeError(s) raised during concurrent store: {runtime_errors}"
    )


@then("the tier service should contain at least {n:d} fragment")
def step_then_contains_at_least(context: Any, n: int) -> None:
    metrics = context.tier_service.get_metrics()
    total = metrics.hot_count + metrics.warm_count + metrics.cold_count
    assert total >= n, f"Expected at least {n} fragments, got {total}"


# ---------------------------------------------------------------------------
# Concurrent store + get
# ---------------------------------------------------------------------------


@given("{n:d} fragments pre-stored in the tier service")
def step_given_pre_stored(context: Any, n: int) -> None:
    for i in range(n):
        frag = _make_fragment(f"pre-{i}", tier=ContextTier.HOT, token_count=5)
        context.tier_service.store(frag)


@when(
    "{s:d} threads concurrently store fragments and "
    "{g:d} threads concurrently get fragments"
)
def step_when_concurrent_store_and_get(context: Any, s: int, g: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()

    def store_worker(thread_idx: int) -> None:
        for i in range(5):
            fid = f"sg-store-t{thread_idx}-f{i}"
            frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5)
            try:
                context.tier_service.store(frag)
            except Exception as exc:
                with lock:
                    errors.append(exc)

    def get_worker(thread_idx: int) -> None:
        for i in range(20):
            fid = f"pre-{i % 20}"
            try:
                context.tier_service.get(fid)
            except Exception as exc:
                with lock:
                    errors.append(exc)

    threads = [threading.Thread(target=store_worker, args=(t,)) for t in range(s)]
    threads += [threading.Thread(target=get_worker, args=(t,)) for t in range(g)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no exception should have been raised during concurrent store and get")
def step_then_no_exc_store_get(context: Any) -> None:
    assert not context.errors, (
        f"Exception(s) raised during concurrent store+get: {context.errors}"
    )


# ---------------------------------------------------------------------------
# Concurrent promote
# ---------------------------------------------------------------------------


@given("{n:d} fragments stored in the cold tier")
def step_given_cold_fragments(context: Any, n: int) -> None:
    context.fragment_ids = []
    for i in range(n):
        fid = f"cold-{i}"
        frag = _make_fragment(fid, tier=ContextTier.COLD, token_count=5)
        context.tier_service.store(frag)
        context.fragment_ids.append(fid)


@when("{n:d} threads concurrently promote all cold fragments")
def step_when_concurrent_promote(context: Any, n: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()
    fids = list(context.fragment_ids)

    def worker() -> None:
        for fid in fids:
            try:
                context.tier_service.promote(fid)
            except Exception as exc:
                with lock:
                    errors.append(exc)

    threads = [threading.Thread(target=worker) for _ in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no exception should have been raised during concurrent promote")
def step_then_no_exc_promote(context: Any) -> None:
    assert not context.errors, (
        f"Exception(s) raised during concurrent promote: {context.errors}"
    )


@then("no fragment should appear in more than one tier")
def step_then_no_duplicate_tier(context: Any) -> None:
    svc = context.tier_service
    # Access private stores directly to verify no duplicates
    hot_ids = set(svc._hot.keys())
    warm_ids = set(svc._warm.keys())
    cold_ids = set(svc._cold.keys())

    hot_warm = hot_ids & warm_ids
    hot_cold = hot_ids & cold_ids
    warm_cold = warm_ids & cold_ids

    assert not hot_warm, f"Fragments in both hot and warm: {hot_warm}"
    assert not hot_cold, f"Fragments in both hot and cold: {hot_cold}"
    assert not warm_cold, f"Fragments in both warm and cold: {warm_cold}"


# ---------------------------------------------------------------------------
# Concurrent demote
# ---------------------------------------------------------------------------


@given("{n:d} fragments stored in the hot tier")
def step_given_hot_fragments(context: Any, n: int) -> None:
    context.fragment_ids = []
    for i in range(n):
        fid = f"hot-{i}"
        frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5)
        context.tier_service.store(frag)
        context.fragment_ids.append(fid)


@when("{n:d} threads concurrently demote all hot fragments")
def step_when_concurrent_demote(context: Any, n: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()
    fids = list(context.fragment_ids)

    def worker() -> None:
        for fid in fids:
            try:
                context.tier_service.demote(fid)
            except Exception as exc:
                with lock:
                    errors.append(exc)

    threads = [threading.Thread(target=worker) for _ in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no exception should have been raised during concurrent demote")
def step_then_no_exc_demote(context: Any) -> None:
    assert not context.errors, (
        f"Exception(s) raised during concurrent demote: {context.errors}"
    )


# ---------------------------------------------------------------------------
# Concurrent evict_lru
# ---------------------------------------------------------------------------


@when("{n:d} threads concurrently evict {k:d} LRU fragment from the hot tier")
def step_when_concurrent_evict(context: Any, n: int, k: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()

    def worker() -> None:
        try:
            context.tier_service.evict_lru(ContextTier.HOT, k)
        except Exception as exc:
            with lock:
                errors.append(exc)

    threads = [threading.Thread(target=worker) for _ in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no exception should have been raised during concurrent evict")
def step_then_no_exc_evict(context: Any) -> None:
    assert not context.errors, (
        f"Exception(s) raised during concurrent evict: {context.errors}"
    )


# ---------------------------------------------------------------------------
# Concurrent enforce_staleness
# ---------------------------------------------------------------------------


@given("{n:d} fragments stored in the hot tier with stale timestamps")
def step_given_stale_hot_fragments(context: Any, n: int) -> None:
    context.fragment_ids = []
    for i in range(n):
        fid = f"stale-hot-{i}"
        frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5, stale=True)
        context.tier_service.store(frag)
        # Manually set stale timestamp after storing
        if fid in context.tier_service._hot:
            context.tier_service._hot[fid].last_accessed = datetime.now(
                tz=UTC
            ) - timedelta(hours=48)
        context.fragment_ids.append(fid)


@when("{n:d} threads concurrently invoke enforce_staleness")
def step_when_concurrent_staleness(context: Any, n: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()

    def worker() -> None:
        try:
            context.tier_service.enforce_staleness()
        except Exception as exc:
            with lock:
                errors.append(exc)

    threads = [threading.Thread(target=worker) for _ in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no exception should have been raised during concurrent staleness enforcement")
def step_then_no_exc_staleness(context: Any) -> None:
    assert not context.errors, (
        f"Exception(s) raised during concurrent enforce_staleness: {context.errors}"
    )


# ---------------------------------------------------------------------------
# Concurrent get_metrics
# ---------------------------------------------------------------------------


@when("{n:d} threads concurrently call get_metrics while {m:d} threads store fragments")
def step_when_concurrent_metrics(context: Any, n: int, m: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()

    def metrics_worker(thread_idx: int) -> None:
        for _ in range(10):
            try:
                context.tier_service.get_metrics()
            except Exception as exc:
                with lock:
                    errors.append(exc)

    def store_worker(thread_idx: int) -> None:
        for i in range(5):
            fid = f"metrics-t{thread_idx}-f{i}"
            frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5)
            try:
                context.tier_service.store(frag)
            except Exception as exc:
                with lock:
                    errors.append(exc)

    threads = [threading.Thread(target=metrics_worker, args=(t,)) for t in range(n)]
    threads += [threading.Thread(target=store_worker, args=(t,)) for t in range(m)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no exception should have been raised during concurrent metrics access")
def step_then_no_exc_metrics(context: Any) -> None:
    assert not context.errors, (
        f"Exception(s) raised during concurrent get_metrics: {context.errors}"
    )


# ---------------------------------------------------------------------------
# Singleton concurrent access
# ---------------------------------------------------------------------------


@when("{n:d} threads each store {k:d} fragments and immediately get them back")
def step_when_concurrent_singleton(context: Any, n: int, k: int) -> None:
    errors: list[Exception] = []
    lock = threading.Lock()

    def worker(thread_idx: int) -> None:
        for i in range(k):
            fid = f"singleton-t{thread_idx}-f{i}"
            frag = _make_fragment(fid, tier=ContextTier.HOT, token_count=5)
            try:
                context.tier_service.store(frag)
                result = context.tier_service.get(fid)
                # result may be None if another thread evicted it; that's OK
                _ = result
            except Exception as exc:
                with lock:
                    errors.append(exc)

    threads = [threading.Thread(target=worker, args=(t,)) for t in range(n)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()

    context.errors = errors


@then("no exception should have been raised during concurrent singleton access")
def step_then_no_exc_singleton(context: Any) -> None:
    assert not context.errors, (
        f"Exception(s) raised during concurrent singleton access: {context.errors}"
    )


@then("the tier service metrics should reflect a non-negative fragment count")
def step_then_non_negative_count(context: Any) -> None:
    metrics = context.tier_service.get_metrics()
    total = metrics.hot_count + metrics.warm_count + metrics.cold_count
    assert total >= 0, f"Negative fragment count: {total}"
