learningcircuit--local-deep-research/tests/web/routes/test_context_overflow_coverage.py

"""
Branch-coverage tests for context_overflow_api.py Flask blueprint.

Covers:
- get_context_overflow_metrics:
  - period "30d" (default), "7d", "3m", "1y", "all"
  - invalid period defaults to "30d"
  - per_page clamping (>500, <1)
  - page clamping (<1)
  - with truncated recent_requests data (truncation_rate > 0 branch)
  - with time series data including truncated entry (ollama_used branch)
  - with model_stats/context_limits populated
  - all_requests pagination (total_count > 0 branch)
  - DB exception returns 500
- get_research_context_overflow:
  - empty token_usage returns early success
  - token_usage with truncated and non-truncated entries
  - research_phase=None → "unknown"
  - no context_limit on any entry
  - DB exception returns 500
"""

from contextlib import contextmanager
from datetime import datetime, timezone
from unittest.mock import MagicMock, patch

import pytest
from flask import Flask

from local_deep_research.web.auth.routes import auth_bp
from local_deep_research.web.routes.context_overflow_api import (
    context_overflow_bp,
)

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

MODULE = "local_deep_research.web.routes.context_overflow_api"
AUTH_DB_MANAGER = "local_deep_research.web.auth.decorators.db_manager"


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


def _create_test_app():
    app = Flask(__name__)
    app.config["SECRET_KEY"] = "test-secret"
    app.config["TESTING"] = True
    app.config["WTF_CSRF_ENABLED"] = False
    app.register_blueprint(auth_bp)
    app.register_blueprint(context_overflow_bp)
    return app


def _make_chainable_query():
    """Create a mock query that supports SQLAlchemy method chaining."""
    query = MagicMock()
    for method in [
        "filter",
        "filter_by",
        "order_by",
        "limit",
        "offset",
        "group_by",
        "with_entities",
        "having",
    ]:
        getattr(query, method).return_value = query
    return query


def _make_overview_row(total=0, with_ctx=0, truncated=0):
    row = MagicMock()
    row.total_requests = total
    row.requests_with_context = with_ctx
    row.truncated_requests = truncated
    return row


def _make_token_summary_row(
    total_requests=0,
    total_tokens=0,
    total_prompt=0,
    total_completion=0,
    avg_prompt=0,
    avg_completion=0,
    max_prompt=0,
):
    row = MagicMock()
    row.total_requests = total_requests
    row.total_tokens = total_tokens
    row.total_prompt_tokens = total_prompt
    row.total_completion_tokens = total_completion
    row.avg_prompt_tokens = avg_prompt
    row.avg_completion_tokens = avg_completion
    row.max_prompt_tokens = max_prompt
    return row


def _make_token_usage(**kwargs):
    tu = MagicMock()
    tu.id = kwargs.get("id", 1)
    tu.timestamp = kwargs.get(
        "timestamp", datetime(2026, 1, 1, tzinfo=timezone.utc)
    )
    tu.research_id = kwargs.get("research_id", "r1")
    tu.model_name = kwargs.get("model_name", "gpt-4")
    tu.model_provider = kwargs.get("model_provider", "openai")
    tu.prompt_tokens = kwargs.get("prompt_tokens", 100)
    tu.completion_tokens = kwargs.get("completion_tokens", 50)
    tu.total_tokens = kwargs.get("total_tokens", 150)
    tu.context_limit = kwargs.get("context_limit", 8192)
    tu.context_truncated = kwargs.get("context_truncated", False)
    tu.tokens_truncated = kwargs.get("tokens_truncated", 0)
    tu.truncation_ratio = kwargs.get("truncation_ratio", 0.0)
    tu.ollama_prompt_eval_count = kwargs.get("ollama_prompt_eval_count", None)
    tu.research_query = kwargs.get("research_query", "test query")
    tu.research_phase = kwargs.get("research_phase", "search")
    tu.calling_function = kwargs.get("calling_function", "test_fn")
    tu.response_time_ms = kwargs.get("response_time_ms", 500)
    tu.content_preview = kwargs.get("content_preview", None)
    return tu


@contextmanager
def _mock_db_session(mock_session):
    yield mock_session


# ---------------------------------------------------------------------------
# Fixtures
# ---------------------------------------------------------------------------


@pytest.fixture()
def app():
    """Minimal Flask app with context_overflow blueprint and mocked auth."""
    with patch(AUTH_DB_MANAGER) as mock_dbm:
        mock_dbm.is_user_connected.return_value = True
        application = _create_test_app()
        yield application


@pytest.fixture()
def client(app):
    return app.test_client()


@pytest.fixture()
def authed_client(client):
    with client.session_transaction() as sess:
        sess["username"] = "testuser"
    with patch(
        f"{MODULE}.SettingsManager",
        return_value=MagicMock(get_setting=MagicMock(return_value=8192)),
    ):
        yield client


# ---------------------------------------------------------------------------
# Shared mock DB setup for the main endpoint
# ---------------------------------------------------------------------------


def _build_main_endpoint_session(
    overview_row=None,
    token_summary_row=None,
    model_token_rows=None,
    phase_rows=None,
    context_limit_rows=None,
    recent_truncated_rows=None,
    time_series_rows=None,
    model_stat_rows=None,
    all_requests_rows=None,
    all_requests_count=0,
    avg_tokens_scalar=0,
):
    """
    Build a mock session whose query chain supports the main endpoint's
    multiple with_entities().first(), .all(), .scalar(), and .count() calls.

    All calls to session.query() return the same single chainable mock because
    mock_session.query.return_value is set to one query object.  The terminal
    methods (.all, .first, .scalar, .count) use side_effect lists to return
    the right value on each successive call.

    Exact call order in the source (get_context_overflow_metrics):

    .first() calls:
      1. overview_row — merged scalar aggregates: counts +
         AVG(tokens_truncated) + token-summary sums/avgs (single query).

    .all() calls:
      1. model_token_query   (query.with_entities(...).group_by(...).all())
      2. phase_query         (query.with_entities(...).group_by(...).all())
      3. context_limits      (session.query(...).filter(...).group_by(...).all())
      4. recent_truncated    (query.filter(...).order_by(...).limit(20).all())
      5. time_series_data    (query.order_by(...).limit(...).all())
      6. model_stats         (session.query(...).filter(...).group_by(...).all())
      7. all_requests_data   (query.order_by(...).offset(...).limit(...).all())

    .count() calls:
      1. all_requests_total  (query.order_by(...).count())
    """
    if overview_row is None:
        overview_row = _make_overview_row()
    if token_summary_row is None:
        token_summary_row = _make_token_summary_row()
    if model_token_rows is None:
        model_token_rows = []
    if phase_rows is None:
        phase_rows = []
    if context_limit_rows is None:
        context_limit_rows = []
    if recent_truncated_rows is None:
        recent_truncated_rows = []
    if time_series_rows is None:
        time_series_rows = []
    if model_stat_rows is None:
        model_stat_rows = []
    if all_requests_rows is None:
        all_requests_rows = []

    query = _make_chainable_query()

    # The route now issues a single .first() call returning a row with all
    # overview + token-summary + avg_tokens_truncated fields combined.
    overview_row.total_tokens = token_summary_row.total_tokens
    overview_row.total_prompt_tokens = token_summary_row.total_prompt_tokens
    overview_row.total_completion_tokens = (
        token_summary_row.total_completion_tokens
    )
    overview_row.avg_prompt_tokens = token_summary_row.avg_prompt_tokens
    overview_row.avg_completion_tokens = token_summary_row.avg_completion_tokens
    overview_row.max_prompt_tokens = token_summary_row.max_prompt_tokens
    overview_row.avg_tokens_truncated = avg_tokens_scalar
    query.first.return_value = overview_row

    # .count() — call 1: all_requests_total
    query.count.side_effect = [all_requests_count]

    # .all() — 7 calls in source order (see docstring above)
    query.all.side_effect = [
        model_token_rows,  # 1. model_token_query
        phase_rows,  # 2. phase_query
        context_limit_rows,  # 3. context_limits
        recent_truncated_rows,  # 4. recent_truncated
        time_series_rows,  # 5. time_series_data
        model_stat_rows,  # 6. model_stats
        all_requests_rows,  # 7. all_requests_data
    ]

    mock_session = MagicMock()
    mock_session.query.return_value = query
    return mock_session


# ---------------------------------------------------------------------------
# Tests: get_context_overflow_metrics
# ---------------------------------------------------------------------------


class TestGetContextOverflowMetrics:
    """Branch coverage for GET /api/context-overflow."""

    def test_default_period_30d_empty_data(self, authed_client):
        """Default period (30d) with empty DB returns success."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "success"
        assert data["overview"]["total_requests"] == 0
        assert data["overview"]["truncation_rate"] == 0
        assert data["pagination"]["per_page"] == 50
        assert data["pagination"]["page"] == 1

    def test_period_all_no_date_filter(self, authed_client):
        """period=all skips start_date calculation branch."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=all")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "success"

    def test_period_7d(self, authed_client):
        """period=7d takes the 7d timedelta branch."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=7d")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "success"

    def test_period_3m(self, authed_client):
        """period=3m takes the 90-day timedelta branch."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=3m")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "success"

    def test_period_1y(self, authed_client):
        """period=1y takes the 365-day timedelta branch."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=1y")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "success"

    def test_invalid_period_defaults_to_30d(self, authed_client):
        """An unrecognised period string is replaced with '30d'."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=bogus")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "success"
        # The response builds normally (no crash), confirming the fallback ran
        assert "pagination" in data

    def test_per_page_clamped_to_max_500(self, authed_client):
        """per_page > 500 is clamped to 500."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?per_page=9999")

        assert resp.status_code == 200
        assert resp.get_json()["pagination"]["per_page"] == 500

    def test_per_page_clamped_to_min_1(self, authed_client):
        """per_page < 1 is clamped to 1."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?per_page=0")

        assert resp.status_code == 200
        assert resp.get_json()["pagination"]["per_page"] == 1

    def test_page_clamped_to_min_1(self, authed_client):
        """page < 1 is clamped to 1."""
        mock_session = _build_main_endpoint_session()

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?page=-3")

        assert resp.status_code == 200
        assert resp.get_json()["pagination"]["page"] == 1

    def test_truncation_rate_calculated_when_context_data_present(
        self, authed_client
    ):
        """When requests_with_context > 0, truncation_rate branch executes."""
        overview_row = _make_overview_row(total=10, with_ctx=8, truncated=4)
        mock_session = _build_main_endpoint_session(
            overview_row=overview_row,
            token_summary_row=_make_token_summary_row(total_requests=10),
            avg_tokens_scalar=200,
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["overview"]["truncation_rate"] == 50.0
        assert data["overview"]["avg_tokens_truncated"] == 200

    def test_with_recent_truncated_rows_in_response(self, authed_client):
        """recent_truncated rows are formatted correctly in the response."""
        trunc_req = _make_token_usage(
            context_truncated=True,
            tokens_truncated=512,
            truncation_ratio=0.0625,
            ollama_prompt_eval_count=None,
        )
        mock_session = _build_main_endpoint_session(
            overview_row=_make_overview_row(total=1, with_ctx=1, truncated=1),
            token_summary_row=_make_token_summary_row(total_requests=1),
            recent_truncated_rows=[trunc_req],
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        assert len(data["recent_truncated"]) == 1
        entry = data["recent_truncated"][0]
        assert entry["tokens_truncated"] == 512
        assert entry["truncation_ratio"] == pytest.approx(0.0625)

    def test_chart_data_with_truncated_entry_uses_ollama_tokens(
        self, authed_client
    ):
        """
        When context_truncated=True and ollama_prompt_eval_count is set,
        original_tokens = ollama_used + tokens_truncated (not prompt_tokens).
        """
        ts_entry = _make_token_usage(
            context_truncated=True,
            tokens_truncated=100,
            ollama_prompt_eval_count=900,
            prompt_tokens=800,
        )
        mock_session = _build_main_endpoint_session(
            overview_row=_make_overview_row(total=1),
            token_summary_row=_make_token_summary_row(total_requests=1),
            time_series_rows=[ts_entry],
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=7d")

        assert resp.status_code == 200
        data = resp.get_json()
        assert len(data["chart_data"]) == 1
        point = data["chart_data"][0]
        # original = ollama_used(900) + tokens_truncated(100)
        assert point["original_prompt_tokens"] == 1000
        assert point["ollama_prompt_tokens"] == 900
        assert point["truncated"] is True

    def test_chart_data_non_truncated_entry_uses_actual_prompt(
        self, authed_client
    ):
        """
        When context_truncated=False, original_tokens == actual_prompt.
        With no ollama count, actual_prompt = prompt_tokens.
        """
        ts_entry = _make_token_usage(
            context_truncated=False,
            tokens_truncated=0,
            ollama_prompt_eval_count=None,
            prompt_tokens=500,
        )
        mock_session = _build_main_endpoint_session(
            overview_row=_make_overview_row(total=1),
            token_summary_row=_make_token_summary_row(total_requests=1),
            time_series_rows=[ts_entry],
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=30d")

        assert resp.status_code == 200
        data = resp.get_json()
        assert len(data["chart_data"]) == 1
        point = data["chart_data"][0]
        assert point["original_prompt_tokens"] == 500
        assert point["truncated"] is False

    def test_longer_period_uses_500_limit_for_time_series(self, authed_client):
        """
        For period in {"3m", "1y"} the time-series query uses .limit(500),
        while "7d" / "30d" uses .limit(1000).  Both paths complete without error.
        """
        mock_session = _build_main_endpoint_session(
            overview_row=_make_overview_row(),
            token_summary_row=_make_token_summary_row(),
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?period=3m")

        assert resp.status_code == 200

    def test_all_requests_pagination_total_pages_calculated(
        self, authed_client
    ):
        """When all_requests_total > 0, total_pages is derived from count."""
        req = _make_token_usage()
        mock_session = _build_main_endpoint_session(
            overview_row=_make_overview_row(total=1),
            token_summary_row=_make_token_summary_row(total_requests=1),
            all_requests_rows=[req],
            all_requests_count=1,
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow?per_page=50&page=1")

        assert resp.status_code == 200
        data = resp.get_json()
        pagination = data["pagination"]
        assert pagination["total_count"] == 1
        assert pagination["total_pages"] == 1
        assert len(data["all_requests"]) == 1
        entry = data["all_requests"][0]
        assert entry["model"] == "gpt-4"
        assert entry["provider"] == "openai"

    def test_model_token_stats_and_phase_breakdown_populated(
        self, authed_client
    ):
        """model_token_stats and phase_breakdown rows are serialised."""
        model_row = MagicMock()
        model_row.model_name = "claude-3"
        model_row.model_provider = "anthropic"
        model_row.total_requests = 5
        model_row.total_tokens = 2000
        model_row.avg_prompt = 300.0
        model_row.max_prompt = 600
        model_row.avg_response_time_ms = 250.0

        phase_row = MagicMock()
        phase_row.research_phase = "synthesis"
        phase_row.count = 3
        phase_row.total_tokens = 900
        phase_row.avg_tokens = 300.0

        mock_session = _build_main_endpoint_session(
            model_token_rows=[model_row],
            phase_rows=[phase_row],
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        assert len(data["model_token_stats"]) == 1
        assert data["model_token_stats"][0]["model"] == "claude-3"
        assert len(data["phase_breakdown"]) == 1
        assert data["phase_breakdown"][0]["phase"] == "synthesis"

    def test_model_stats_with_none_avg_context_limit(self, authed_client):
        """model_stats entry with avg_context_limit=None produces None in output."""
        stat = MagicMock()
        stat.model_name = "llama"
        stat.model_provider = "ollama"
        stat.total_requests = 2
        stat.truncated_count = 0
        stat.avg_context_limit = None

        mock_session = _build_main_endpoint_session(
            model_stat_rows=[stat],
        )

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        assert len(data["model_stats"]) == 1
        assert data["model_stats"][0]["avg_context_limit"] is None

    def test_model_stats_truncation_rate_when_total_requests_zero(
        self, authed_client
    ):
        """model_stats entry with total_requests=0 yields truncation_rate=0."""
        stat = MagicMock()
        stat.model_name = "mystery"
        stat.model_provider = "unknown"
        stat.total_requests = 0
        stat.truncated_count = 0
        stat.avg_context_limit = 4096

        mock_session = _build_main_endpoint_session(model_stat_rows=[stat])

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["model_stats"][0]["truncation_rate"] == 0

    def test_db_exception_returns_500(self, authed_client):
        """When get_user_db_session raises, endpoint returns 500."""

        @contextmanager
        def _exploding(*args, **kwargs):
            raise RuntimeError("simulated DB failure")
            yield  # pragma: no cover

        with patch(f"{MODULE}.get_user_db_session", side_effect=_exploding):
            resp = authed_client.get("/api/context-overflow")

        assert resp.status_code == 500
        data = resp.get_json()
        assert data["status"] == "error"
        assert "context overflow metrics" in data["message"]

    def test_unauthenticated_returns_401(self, client):
        """Without a session the endpoint rejects the request."""
        resp = client.get("/api/context-overflow")
        assert resp.status_code in (401, 302)


# ---------------------------------------------------------------------------
# Tests: get_research_context_overflow
# ---------------------------------------------------------------------------


class TestGetResearchContextOverflow:
    """Branch coverage for GET /api/research/<id>/context-overflow."""

    def test_empty_token_usage_returns_early_empty_response(
        self, authed_client
    ):
        """No rows for research_id → early return with zero overview."""
        mock_session = MagicMock()
        query = _make_chainable_query()
        query.all.return_value = []
        mock_session.query.return_value = query

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get(
                "/api/research/nonexistent/context-overflow"
            )

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["status"] == "success"
        assert data["data"]["overview"]["total_requests"] == 0
        assert data["data"]["overview"]["total_tokens"] == 0
        assert data["data"]["overview"]["truncation_occurred"] is False
        assert data["data"]["requests"] == []

    def test_with_mixed_truncated_and_normal_usage(self, authed_client):
        """Token usage with both truncated and non-truncated records."""
        normal = _make_token_usage(
            id=1,
            context_truncated=False,
            tokens_truncated=0,
            research_phase="search",
            total_tokens=200,
            prompt_tokens=150,
            completion_tokens=50,
            context_limit=8192,
        )
        truncated = _make_token_usage(
            id=2,
            context_truncated=True,
            tokens_truncated=300,
            research_phase="search",
            total_tokens=500,
            prompt_tokens=450,
            completion_tokens=50,
            context_limit=8192,
        )
        mock_session = MagicMock()
        query = _make_chainable_query()
        query.all.return_value = [normal, truncated]
        mock_session.query.return_value = query

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/research/r1/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        overview = data["data"]["overview"]
        assert overview["total_requests"] == 2
        assert overview["total_tokens"] == 700
        assert overview["truncation_occurred"] is True
        assert overview["truncated_count"] == 1
        assert overview["tokens_lost"] == 300
        assert overview["context_limit"] == 8192
        assert overview["max_tokens_used"] == 450
        assert data["data"]["model"] == "gpt-4"
        assert data["data"]["provider"] == "openai"

    def test_phase_stats_accumulation(self, authed_client):
        """Phase stats correctly accumulate counts and token sums per phase."""
        entry_a = _make_token_usage(
            research_phase="search",
            total_tokens=100,
            prompt_tokens=80,
            completion_tokens=20,
            context_truncated=False,
            tokens_truncated=0,
        )
        entry_b = _make_token_usage(
            research_phase="search",
            total_tokens=120,
            prompt_tokens=90,
            completion_tokens=30,
            context_truncated=True,
            tokens_truncated=50,
        )
        entry_c = _make_token_usage(
            research_phase="synthesis",
            total_tokens=200,
            prompt_tokens=160,
            completion_tokens=40,
            context_truncated=False,
            tokens_truncated=0,
        )
        mock_session = MagicMock()
        query = _make_chainable_query()
        query.all.return_value = [entry_a, entry_b, entry_c]
        mock_session.query.return_value = query

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/research/r1/context-overflow")

        assert resp.status_code == 200
        phase_stats = resp.get_json()["data"]["phase_stats"]
        assert "search" in phase_stats
        assert "synthesis" in phase_stats
        assert phase_stats["search"]["count"] == 2
        assert phase_stats["search"]["truncated_count"] == 1
        assert phase_stats["search"]["total_tokens"] == 220
        assert phase_stats["synthesis"]["count"] == 1
        assert phase_stats["synthesis"]["truncated_count"] == 0

    def test_none_research_phase_maps_to_unknown(self, authed_client):
        """research_phase=None is bucketed under 'unknown'."""
        entry = _make_token_usage(research_phase=None)
        mock_session = MagicMock()
        query = _make_chainable_query()
        query.all.return_value = [entry]
        mock_session.query.return_value = query

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/research/r1/context-overflow")

        assert resp.status_code == 200
        phase_stats = resp.get_json()["data"]["phase_stats"]
        assert "unknown" in phase_stats

    def test_no_context_limit_on_any_entry(self, authed_client):
        """When no entry has a context_limit, overview.context_limit is None."""
        entry = _make_token_usage(context_limit=None)
        mock_session = MagicMock()
        query = _make_chainable_query()
        query.all.return_value = [entry]
        mock_session.query.return_value = query

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/research/r1/context-overflow")

        assert resp.status_code == 200
        data = resp.get_json()
        assert data["data"]["overview"]["context_limit"] is None

    def test_requests_list_contains_correct_fields(self, authed_client):
        """Each entry in requests[] includes all expected keys."""
        entry = _make_token_usage(
            ollama_prompt_eval_count=42,
            calling_function="my_fn",
            response_time_ms=300,
            tokens_truncated=0,
            context_truncated=False,
        )
        mock_session = MagicMock()
        query = _make_chainable_query()
        query.all.return_value = [entry]
        mock_session.query.return_value = query

        with patch(
            f"{MODULE}.get_user_db_session",
            return_value=_mock_db_session(mock_session),
        ):
            resp = authed_client.get("/api/research/r1/context-overflow")

        assert resp.status_code == 200
        req_entry = resp.get_json()["data"]["requests"][0]
        for key in [
            "timestamp",
            "phase",
            "prompt_tokens",
            "completion_tokens",
            "total_tokens",
            "context_limit",
            "context_truncated",
            "tokens_truncated",
            "ollama_prompt_eval_count",
            "calling_function",
            "response_time_ms",
        ]:
            assert key in req_entry, f"missing key: {key}"
        assert req_entry["ollama_prompt_eval_count"] == 42
        assert req_entry["calling_function"] == "my_fn"
        assert req_entry["context_truncated"] is False

    def test_db_exception_returns_500(self, authed_client):
        """When get_user_db_session raises, endpoint returns 500."""

        @contextmanager
        def _exploding(*args, **kwargs):
            raise RuntimeError("simulated DB failure")
            yield  # pragma: no cover

        with patch(f"{MODULE}.get_user_db_session", side_effect=_exploding):
            resp = authed_client.get("/api/research/r1/context-overflow")

        assert resp.status_code == 500
        data = resp.get_json()
        assert data["status"] == "error"
        assert "context overflow data" in data["message"]

    def test_unauthenticated_returns_401(self, client):
        """Without a session the research endpoint rejects the request."""
        resp = client.get("/api/research/r1/context-overflow")
        assert resp.status_code in (401, 302)