microdao-daarion/tests/test_stepan_doc_mode_hardening_v36.py

"""
tests/test_stepan_doc_mode_hardening_v36.py

v3.6 Doc Mode Hardening:
  - Cooldown blocks implicit doc re-activate (Rule 1)
  - Explicit doc token bypasses cooldown (Rule 3)
  - Fact signal allows implicit doc (Rule 2)
  - No fact signal + no explicit → clarifier
  - Auto-clear sets cooldown (cooldown_set)
  - build_mode_clarifier (URL, vision, fact, neutral)
  - detect_context_signals correctness
  - is_doc_focus_cooldown_active TTL
  - /doc status shows cooldown_left
"""
import time
import pytest


# ─── helpers для ізольованого тестування без crewai ──────────────────────────
from crews.agromatrix_crew.doc_focus import (
    _is_doc_question,
    _detect_domain,
    detect_context_signals,
    build_mode_clarifier,
    handle_doc_focus,
)
from crews.agromatrix_crew.session_context import (
    _default_session,
    is_doc_focus_cooldown_active,
    is_doc_focus_active,
    DOC_FOCUS_TTL,
    DOC_FOCUS_COOLDOWN_S,
)


# ─── 1. detect_context_signals ────────────────────────────────────────────────

class TestDetectContextSignals:
    def test_empty_text(self):
        s = detect_context_signals("")
        assert not any(s.values()), "Empty text should produce all-False signals"

    def test_explicit_doc_token(self):
        s = detect_context_signals("по звіту що таке прибуток?")
        assert s["has_explicit_doc_token"]
        assert s["has_doc_trigger"]

    def test_url_detected(self):
        s = detect_context_signals("дивись https://example.com")
        assert s["has_url"]
        assert not s["has_explicit_doc_token"]

    def test_vision_trigger(self):
        s = detect_context_signals("на листі плями чорні")
        assert s["has_vision_trigger"]
        assert not s["has_explicit_doc_token"]

    def test_fact_signal_units(self):
        s = detect_context_signals("скільки грн/га вийшло?")
        assert s["has_fact_signal"]

    def test_fact_signal_words(self):
        s = detect_context_signals("що з витратами на добрива?")
        assert s["has_fact_signal"]

    def test_no_signals(self):
        s = detect_context_signals("привіт, як справи?")
        assert not s["has_explicit_doc_token"]
        assert not s["has_url"]
        assert not s["has_fact_signal"]

    def test_fact_units_number(self):
        s = detect_context_signals("4500 грн на гектар")
        assert s["has_fact_signal"]


# ─── 2. build_mode_clarifier ──────────────────────────────────────────────────

class TestBuildModeClarifier:
    def test_url_clarifier(self):
        c = build_mode_clarifier("ось посилання https://site.com")
        assert "посилання" in c.lower()
        assert "?" in c
        assert "!" not in c

    def test_vision_clarifier(self):
        c = build_mode_clarifier("на листі плями є")
        assert "фото" in c.lower() or "рослин" in c.lower()

    def test_fact_clarifier(self):
        c = build_mode_clarifier("а що з витратами?")
        assert "цифри" in c.lower() or "звіт" in c.lower()

    def test_neutral_clarifier(self):
        c = build_mode_clarifier("що там?")
        assert "звіт" in c.lower() or "інше" in c.lower()

    def test_no_exclamation(self):
        for text in ["https://x.com", "листя плями", "витрати?", "привіт"]:
            assert "!" not in build_mode_clarifier(text)

    def test_exactly_one_question_mark(self):
        for text in ["https://x.com", "листя плями", "витрати?", "привіт"]:
            assert build_mode_clarifier(text).count("?") == 1


# ─── 3. is_doc_focus_cooldown_active ─────────────────────────────────────────

class TestCooldownHelper:
    def test_inactive_by_default(self):
        s = _default_session()
        assert not is_doc_focus_cooldown_active(s)

    def test_active_when_in_future(self):
        now = time.time()
        s = _default_session()
        s["doc_focus_cooldown_until"] = now + 60
        assert is_doc_focus_cooldown_active(s, now)

    def test_expired(self):
        now = time.time()
        s = _default_session()
        s["doc_focus_cooldown_until"] = now - 1
        assert not is_doc_focus_cooldown_active(s, now)

    def test_zero_is_inactive(self):
        s = _default_session()
        s["doc_focus_cooldown_until"] = 0.0
        assert not is_doc_focus_cooldown_active(s)

    def test_fail_safe_on_garbage(self):
        s = _default_session()
        s["doc_focus_cooldown_until"] = "broken"
        # Should return False without exception
        result = is_doc_focus_cooldown_active(s)
        assert result is False


# ─── 4. DOC_FOCUS_COOLDOWN_S constant ────────────────────────────────────────

class TestCooldownConstants:
    def test_cooldown_120s(self):
        assert DOC_FOCUS_COOLDOWN_S == 120.0

    def test_ttl_600s(self):
        assert DOC_FOCUS_TTL == 600.0


# ─── 5. Gating симуляція (без LLM/HTTP) ──────────────────────────────────────

def _simulate_gating(text: str, session: dict, now: float) -> tuple[str, str | None]:
    """
    Спрощена симуляція логіки context_mode з run.py.
    Повертає (context_mode, denied_reason | None).
    """
    signals = detect_context_signals(text)
    domain = _detect_domain(text)
    focus_active = is_doc_focus_active(session, now)
    cooldown_active = is_doc_focus_cooldown_active(session, now)

    if domain == "doc":
        is_explicit = signals["has_explicit_doc_token"]
        # Rule 1: cooldown blocks implicit
        if cooldown_active and not is_explicit:
            return "general", "cooldown"
        # Rule 2: implicit → needs fact_signal
        if not is_explicit:
            if signals["has_fact_signal"]:
                return "doc", None
            return "general", "no_fact_signal"
        # Rule 3: explicit always allowed
        return "doc", None

    return "general", None


class TestGatingRules:
    """Rule 1/2/3 з v3.6."""

    def test_cooldown_blocks_implicit_doc(self):
        now = time.time()
        s = _default_session()
        s["doc_focus"] = True
        s["doc_focus_ts"] = now
        s["doc_focus_cooldown_until"] = now + 120
        # "документ що там" — загальний тригер (_DOC_QUESTION_RE),
        # але НЕ explicit ("в документі" потрапляє в explicit regex).
        # Тому використовуємо текст без конструкції "в/у документі":
        mode, reason = _simulate_gating("документ що там", s, now)
        assert mode == "general"
        assert reason == "cooldown"

    def test_explicit_bypasses_cooldown(self):
        now = time.time()
        s = _default_session()
        s["doc_focus"] = True
        s["doc_focus_ts"] = now
        s["doc_focus_cooldown_until"] = now + 120
        mode, reason = _simulate_gating("по звіту прибуток?", s, now)
        assert mode == "doc"
        assert reason is None

    def test_no_explicit_no_fact_signal_denied(self):
        now = time.time()
        s = _default_session()
        s["doc_facts"] = {}
        mode, reason = _simulate_gating("а що там у документі?", s, now)
        # domain=doc (загальний тригер), але signals.has_fact_signal=False → denied
        # Може бути mode=doc якщо has_doc_trigger, перевіряємо логіку
        signals = detect_context_signals("а що там у документі?")
        if not signals["has_fact_signal"] and not signals["has_explicit_doc_token"]:
            assert mode == "general"
            assert reason == "no_fact_signal"

    def test_fact_signal_allows_implicit_doc(self):
        now = time.time()
        s = _default_session()
        s["doc_focus"] = True
        s["doc_focus_ts"] = now
        s["doc_focus_cooldown_until"] = 0.0  # cooldown inactive
        mode, reason = _simulate_gating("скільки грн/га витрат?", s, now)
        # has_fact_signal=True і domain=?
        signals = detect_context_signals("скільки грн/га витрат?")
        domain = _detect_domain("скільки грн/га витрат?")
        if domain == "doc":
            assert mode == "doc"
        # Якщо domain != doc (загальний текст з units) — general, ok теж
        assert reason is None or reason == "no_fact_signal"

    def test_url_domain_goes_general(self):
        now = time.time()
        s = _default_session()
        mode, reason = _simulate_gating("дивись https://example.com", s, now)
        assert mode == "general"

    def test_empty_text_general(self):
        now = time.time()
        s = _default_session()
        mode, reason = _simulate_gating("", s, now)
        assert mode == "general"


# ─── 6. Auto-clear sets cooldown симуляція ────────────────────────────────────

def _simulate_auto_clear(session: dict, domain: str, now: float):
    """Симуляція auto-clear логіки."""
    focus_active = is_doc_focus_active(session, now)
    if focus_active and domain in ("vision", "web"):
        session["doc_focus"] = False
        session["doc_focus_ts"] = 0.0
        session["doc_focus_cooldown_until"] = now + DOC_FOCUS_COOLDOWN_S
        return True  # cleared
    return False


class TestAutoClearCooldown:
    def test_web_domain_sets_cooldown(self):
        now = time.time()
        s = _default_session()
        s["doc_focus"] = True
        s["doc_focus_ts"] = now
        cleared = _simulate_auto_clear(s, "web", now)
        assert cleared
        assert not s["doc_focus"]
        assert is_doc_focus_cooldown_active(s, now)
        assert s["doc_focus_cooldown_until"] == pytest.approx(now + 120, abs=1)

    def test_vision_domain_sets_cooldown(self):
        now = time.time()
        s = _default_session()
        s["doc_focus"] = True
        s["doc_focus_ts"] = now
        _simulate_auto_clear(s, "vision", now)
        assert is_doc_focus_cooldown_active(s, now)

    def test_general_domain_no_clear(self):
        now = time.time()
        s = _default_session()
        s["doc_focus"] = True
        s["doc_focus_ts"] = now
        cleared = _simulate_auto_clear(s, "general", now)
        assert not cleared
        assert s["doc_focus"]

    def test_cooldown_expires_after_120s(self):
        now = time.time()
        s = _default_session()
        s["doc_focus_cooldown_until"] = now - 1  # вже закінчився
        assert not is_doc_focus_cooldown_active(s, now)


# ─── 7. /doc status shows cooldown ────────────────────────────────────────────

class TestDocFocusHandlerCooldown:
    def test_status_no_cooldown(self, monkeypatch):
        """Status при відсутньому cooldown."""
        from crews.agromatrix_crew.session_context import _STORE
        chat_id = "test_v36_status_no_cooldown"
        _STORE[chat_id] = _default_session()
        result = handle_doc_focus("status", chat_id)
        assert result["ok"]
        assert "cooldown" not in result["message"]

    def test_status_with_cooldown(self, monkeypatch):
        """Status показує cooldown_left якщо cooldown активний."""
        from crews.agromatrix_crew.session_context import _STORE
        now = time.time()
        chat_id = "test_v36_status_with_cooldown"
        s = _default_session()
        s["doc_focus_cooldown_until"] = now + 60
        s["updated_at"] = now  # щоб сесія не вважалась протухлою
        _STORE[chat_id] = s
        result = handle_doc_focus("status", chat_id)
        assert result["ok"]
        assert "cooldown" in result["message"]

    def test_doc_on_resets_cooldown(self):
        """'/doc on' скидає cooldown."""
        from crews.agromatrix_crew.session_context import _STORE
        now = time.time()
        chat_id = "test_v36_on_resets_cooldown"
        s = _default_session()
        s["doc_focus_cooldown_until"] = now + 100
        s["updated_at"] = now
        _STORE[chat_id] = s
        handle_doc_focus("on", chat_id)
        updated = _STORE[chat_id]
        assert updated["doc_focus"] is True
        assert updated.get("doc_focus_cooldown_until", 0.0) == 0.0


# ─── 8. Default session has new field ─────────────────────────────────────────

class TestDefaultSession:
    def test_has_cooldown_field(self):
        s = _default_session()
        assert "doc_focus_cooldown_until" in s
        assert s["doc_focus_cooldown_until"] == 0.0

    def test_has_doc_focus(self):
        s = _default_session()
        assert "doc_focus" in s
        assert s["doc_focus"] is False