backend/tests/test_agentic.py

"""Unit tests for the agentic orchestrator native (google.genai) fallback."""
import os
import sys

os.environ.setdefault("CEPHEUS_CLOUD", "1")

BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if BACKEND_DIR not in sys.path:
    sys.path.insert(0, BACKEND_DIR)

import agentic_orchestrator as orch  # noqa: E402
import gemini_config as gc  # noqa: E402


def test_tool_registry_is_complete():
    names = set(orch._TOOL_MAP)
    for expected in (
        "get_current_alerts",
        "get_crowd_statistics",
        "search_person_in_camera_feeds",
        "create_tactical_issue",
        "get_interaction_graph",
        "locate_nearest_emergency_services",
        "list_enrolled_faces",
        "set_ai_model_state",
        "get_active_issues",
        "get_platform_overview",
        "broadcast_alert",
        "set_signage_state",
        "update_issue",
        "get_live_camera_faces",
    ):
        assert expected in names
    assert len(orch._ALL_TOOLS) >= 18
    # Every tool maps back to a callable
    assert all(callable(fn) for fn in orch._TOOL_MAP.values())


def test_parse_retry_delay_extracts_seconds():
    msg = "429 RESOURCE_EXHAUSTED ... 'retryDelay': '58s' ..."
    assert gc.parse_retry_delay(msg) >= 58
    # Falls back to a sane default when no delay is present
    assert gc.parse_retry_delay("no retry hint here") == 5.0


def test_gemini_model_tiers():
    assert gc.get_model("default") == "gemini-3.5-flash"
    assert gc.get_model("pro") == "gemini-3.1-pro-preview"
    assert gc.get_model("lite") == "gemini-3.1-flash-lite"
    assert gc.fallback_chain("default") == ["gemini-3.5-flash", "gemini-3.1-flash-lite"]
    # Pro tier degrades Pro -> Flash -> Flash-Lite (deduped, ordered).
    assert gc.fallback_chain("pro") == [
        "gemini-3.1-pro-preview",
        "gemini-3.5-flash",
        "gemini-3.1-flash-lite",
    ]


def test_generate_with_fallback_degrades_on_rate_limit():
    """A quota-exhausted primary model should transparently fall through to a working one."""
    calls = []

    class _RateLimitError(Exception):
        pass

    class _FakeModels:
        def generate_content(self, model, contents, config):
            calls.append(model)
            if model != "gemini-3.1-flash-lite":
                raise _RateLimitError("429 RESOURCE_EXHAUSTED")
            return f"ok:{model}"

    class _FakeClient:
        models = _FakeModels()

    result = gc.generate_with_fallback(_FakeClient(), tier="pro", contents="hi", config=None)
    assert result == "ok:gemini-3.1-flash-lite"
    # It tried Pro and Flash before succeeding on Flash-Lite.
    assert calls[-1] == "gemini-3.1-flash-lite"
    assert "gemini-3.1-pro-preview" in calls


def test_search_person_is_deterministic_and_honest():
    """Search must reflect real injected data, not guesses, and be repeatable."""
    detections = [{"name": "Vidit", "camId": "cam-02", "confidence": 0.9, "timestamp": "10:00:00"}]
    orch.inject_context(
        alerts=[], sos_events=[], vision_engine=None, rooms=[],
        detections_history=detections, issues=[],
    )
    # Known recent sighting -> RECENTLY_SEEN but not live (found=false).
    r1 = orch.search_person_in_camera_feeds("Vidit")
    r2 = orch.search_person_in_camera_feeds("Vidit")
    assert r1["found"] is False and r1 == r2
    assert r1["status"] == "RECENTLY_SEEN"
    assert r1["last_seen_camera"] == "cam-02"
    # Absent person -> not found, no fabrication.
    miss = orch.search_person_in_camera_feeds("Nobody")
    assert miss["found"] is False
    assert miss["status"] == "NOT_FOUND"


def test_get_live_camera_faces_uses_face_results_only():
    class FakeVE:
        face_results = {
            "cam-01": [{"name": "MK", "confidence": 0.9, "found": True}],
        }
        browser_feeds = {"cam-01": {"status": "ACTIVE", "last_seen": "now", "source": "browser"}}
        camera_indices = {"cam-01": 0}

        def get_ai_status(self):
            return {"facial": True}

    ve = FakeVE()
    orch.inject_context(
        alerts=[], sos_events=[], vision_engine=ve, rooms=[],
        detections_history=[{"name": "Vidit", "camId": "cam-01"}], issues=[],
    )
    live = orch.get_live_camera_faces()
    assert live["total_live_faces"] == 1
    assert live["live_persons"][0]["name"] == "MK"
    assert live["live_persons"][0]["status"] == "LIVE_DETECTED"
    assert live["active_feed_count"] >= 1
    hist = orch.get_known_persons()
    assert hist["data_type"] == "SIGHTING_HISTORY"
    assert len(hist["persons"]) == 1
    assert hist["persons"][0]["name"] == "Vidit"


def test_friendly_model_error_messages():
    quota = orch._friendly_model_error(Exception("429 RESOURCE_EXHAUSTED quota"))
    assert "rate-limited" in quota.lower()

    auth = orch._friendly_model_error(Exception("Invalid API key provided"))
    assert "authentication" in auth.lower()

    generic = orch._friendly_model_error(Exception("something else broke"))
    assert "failed" in generic.lower()


def test_create_tactical_issue_deduplicates_same_subject():
    issues = []
    orch.inject_context(
        alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues
    )
    orch._ctx["allow_issue_creation"] = True
    first = orch.create_tactical_issue(
        "Missing Person Database Match: MK",
        "Database match for MK (Confidence: 0.744)",
        "HIGH",
        '{"person_name":"MK","confidence":0.744}',
    )
    second = orch.create_tactical_issue(
        "Database Match: MK (Pending Live Verification)",
        "Database match for MK (Confidence: 0.763)",
        "HIGH",
        '{"person_name":"MK","confidence":0.763}',
    )
    assert first["success"] is True
    assert second["success"] is True
    assert second.get("deduplicated") is True
    assert len(issues) == 1
    assert issues[0]["metadata"].get("confidence") == 0.763


def test_create_tactical_issue_blocked_without_explicit_intent():
    issues = []
    orch.inject_context(
        alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues
    )
    orch._ctx["allow_issue_creation"] = False
    result = orch.create_tactical_issue(
        "Missing Person Database Match: MK",
        "Database match for MK",
        "HIGH",
        '{"person_name":"MK","confidence":0.744}',
    )
    assert result["success"] is False
    assert result.get("blocked") is True
    assert len(issues) == 0


def test_prompt_requests_issue_creation():
    assert orch.prompt_requests_issue_creation("Search MK and raise an issue if found")
    assert orch.prompt_requests_issue_creation("Please create an issue for this match")
    assert not orch.prompt_requests_issue_creation("Search the live camera feed for enrolled faces")
    assert not orch.prompt_requests_issue_creation("Run a face scan on the current camera feed")


def test_create_tactical_issue_uses_injected_db():
    issues = []
    orch.inject_context(
        alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues
    )
    orch._ctx["allow_issue_creation"] = True
    result = orch.create_tactical_issue("Test", "A test issue", "HIGH", "{}")
    assert result["success"] is True
    assert issues and issues[0]["title"] == "Test"


def test_platform_overview_and_issue_update():
    issues = [{"id": "ISS-1", "title": "X", "status": "ONGOING", "progress": 0, "desc": "d"}]
    orch.inject_context(
        alerts=[{"id": "a1"}], sos_events=[], vision_engine=None, rooms=[],
        detections_history=[{"name": "Vidit"}], issues=issues,
        incident_logs=[{"msg": "boot"}], signage_state={"s6": True, "s1": False},
    )
    ov = orch.get_platform_overview()
    assert ov["active_alerts"] == 1
    assert ov["active_issues"] == 1
    assert ov["live_face_count"] == 0
    assert ov["active_signage"] == ["s6"]

    # update_issue mutates the injected list deterministically.
    res = orch.update_issue("ISS-1", status="RESOLVED", progress=100, note="done")
    assert res["success"] is True
    assert issues[0]["status"] == "RESOLVED" and issues[0]["progress"] == 100

    # Action tools degrade gracefully when no handler is wired.
    orch._action_handlers.clear()
    alert_res = orch.broadcast_alert("fire", "Smoke in corridor", "Corridor B", "critical")
    assert alert_res["success"] is False


def test_action_handlers_invoked():
    captured = {}
    orch.set_action_handlers(
        broadcast_alert=lambda t, m, l, s: captured.update({"alert": (t, m, l, s)}) or {"id": "x"},
        set_signage=lambda sid, active: captured.update({"signage": (sid, active)}),
    )
    orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=[])
    assert orch.broadcast_alert("medical", "Patient down", "Hall", "high")["success"] is True
    assert captured["alert"] == ("medical", "Patient down", "Hall", "high")
    assert orch.set_signage_state("s3", True)["success"] is True
    assert captured["signage"] == ("s3", True)
    orch._action_handlers.clear()


def test_broadcast_alert_message_location_order():
    captured = {}
    orch.set_action_handlers(
        broadcast_alert=lambda t, m, l, s: captured.update({"alert": (t, m, l, s)}) or {"id": "x"},
    )
    orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=[])
    orch.broadcast_alert("fire", "Smoke in corridor", "Building A", "critical")
    assert captured["alert"][1] == "Smoke in corridor"
    assert captured["alert"][2] == "Building A"
    orch._action_handlers.clear()


def test_dispatch_personnel_requires_linked_issue():
    issues = [{"id": "ISS-OTHER", "title": "Unrelated", "desc": "elsewhere", "metadata": {}}]
    orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues)
    res = orch.dispatch_personnel_to_camera("cam-99", 2)
    assert res["success"] is False
    assert "create_tactical_issue" in res["message"]


def test_adk_tool_registry_parity():
    native_names = set(orch._TOOL_MAP)
    adk_names = orch.adk_tool_names()
    missing = native_names - adk_names
    assert not missing, f"ADK agents missing tools: {sorted(missing)}"


def test_decode_data_url_image():
    raw = b"fake-image-bytes"
    import base64 as b64
    data_url = "data:image/png;base64," + b64.b64encode(raw).decode()
    decoded, mime = orch._decode_data_url_image(data_url)
    assert decoded == raw
    assert mime == "image/png"


def test_update_issue_accepts_float_progress():
    issues = [{"id": "ISS-1", "title": "X", "status": "ONGOING", "progress": 0, "desc": "d"}]
    orch.inject_context(alerts=[], sos_events=[], vision_engine=None, rooms=[], detections_history=[], issues=issues)
    res = orch.update_issue("ISS-1", progress=50.6)
    assert res["success"] is True
    assert issues[0]["progress"] == 51


def test_evacuation_plan_marked_unverified():
    plan = orch.generate_evacuation_plan("PLATFORM 1", 100, "fire")
    assert plan.get("verified") is False
    assert "TEMPLATE" in plan.get("note", "")