learningcircuit--local-deep-research/tests/chat/test_chat_concurrency_guard.py

# allow: no-sut-import — black-box HTTP test; drives real routes through the Flask test client
"""Tests for the per-session concurrency guard on POST /api/chat/sessions/<id>/messages.

Chat is sequential by design: at most one in-flight research per session.
A second send while a research is IN_PROGRESS must return 409 with the
active research_id.
"""

import json
from unittest.mock import patch


def _create_session(client, query="Test"):
    resp = client.post(
        "/api/chat/sessions",
        json={"initial_query": query},
        content_type="application/json",
    )
    assert resp.status_code == 200, resp.data
    return json.loads(resp.data)["session_id"]


class TestPerSessionConcurrencyGuard:
    """Per-session 409 guard tests."""

    def test_second_send_while_in_progress_returns_409(
        self, authenticated_client
    ):
        """A second send while the first research is still IN_PROGRESS
        must return 409 with the active research_id."""
        session_id = _create_session(authenticated_client)

        with patch("local_deep_research.chat.routes.start_research_process"):
            first = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": "first", "trigger_research": True},
                content_type="application/json",
            )
            assert first.status_code == 200
            first_rid = json.loads(first.data)["research_id"]
            assert first_rid

            # Second send: research_id from first is still IN_PROGRESS
            # because start_research_process was mocked (no thread ran),
            # so the row remains at status=in_progress.
            second = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": "second", "trigger_research": True},
                content_type="application/json",
            )

            assert second.status_code == 409, second.data
            data = json.loads(second.data)
            assert data["success"] is False
            assert "in progress" in data["error"].lower()
            assert data.get("active_research_id") == first_rid

    def test_second_send_after_terminate_succeeds(self, authenticated_client):
        """Once the first research is terminated (status leaves IN_PROGRESS),
        a second send is accepted normally — i.e. the guard releases."""
        session_id = _create_session(authenticated_client)

        with patch("local_deep_research.chat.routes.start_research_process"):
            first = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": "first", "trigger_research": True},
                content_type="application/json",
            )
            assert first.status_code == 200
            first_rid = json.loads(first.data)["research_id"]

        # Terminate via the existing endpoint — this is the real user flow:
        # click Stop, then retry. The endpoint flips status off IN_PROGRESS
        # without requiring the worker thread to have started.
        term_resp = authenticated_client.post(f"/api/terminate/{first_rid}")
        assert term_resp.status_code == 200, term_resp.data

        with patch("local_deep_research.chat.routes.start_research_process"):
            second = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": "second", "trigger_research": True},
                content_type="application/json",
            )
            assert second.status_code == 200, second.data
            assert json.loads(second.data)["research_id"] != first_rid

    def test_guard_does_not_apply_when_trigger_research_false(
        self, authenticated_client
    ):
        """A trigger_research=False send must succeed even when a research
        is in flight — non-research messages don't conflict."""
        session_id = _create_session(authenticated_client)

        with patch("local_deep_research.chat.routes.start_research_process"):
            first = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": "first", "trigger_research": True},
                content_type="application/json",
            )
            assert first.status_code == 200

            second = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": "no-research", "trigger_research": False},
                content_type="application/json",
            )
            assert second.status_code == 200, second.data
            assert json.loads(second.data)["research_id"] is None

    def test_guard_isolates_per_session(self, authenticated_client):
        """An in-flight research on session A must not block sends on
        session B for the same user."""
        session_a = _create_session(authenticated_client, query="topic A")
        session_b = _create_session(authenticated_client, query="topic B")

        with patch("local_deep_research.chat.routes.start_research_process"):
            ra = authenticated_client.post(
                f"/api/chat/sessions/{session_a}/messages",
                json={"content": "first on A", "trigger_research": True},
                content_type="application/json",
            )
            assert ra.status_code == 200

            # B should be unaffected.
            rb = authenticated_client.post(
                f"/api/chat/sessions/{session_b}/messages",
                json={"content": "first on B", "trigger_research": True},
                content_type="application/json",
            )
            assert rb.status_code == 200, rb.data

    def test_blocked_send_does_not_create_orphan_user_message(
        self, authenticated_client
    ):
        """When the guard returns 409, the rejected user message must NOT
        be in chat_messages — otherwise we'd accumulate orphan rows."""
        session_id = _create_session(authenticated_client)

        with patch("local_deep_research.chat.routes.start_research_process"):
            authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={"content": "first", "trigger_research": True},
                content_type="application/json",
            )

            blocked = authenticated_client.post(
                f"/api/chat/sessions/{session_id}/messages",
                json={
                    "content": "should-be-rejected",
                    "trigger_research": True,
                },
                content_type="application/json",
            )
            assert blocked.status_code == 409

        # Read messages back; only the first user message should exist.
        list_resp = authenticated_client.get(
            f"/api/chat/sessions/{session_id}/messages"
        )
        assert list_resp.status_code == 200
        msgs = json.loads(list_resp.data)["messages"]
        user_contents = [m["content"] for m in msgs if m["role"] == "user"]
        assert "should-be-rejected" not in user_contents