FabledScribe/tests/test_services_dedup.py

"""Unit tests for the write-time near-duplicate gate (services/dedup.py)."""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from scribe.services.dedup import (
    DuplicateMatch,
    duplicate_response,
    find_duplicate_note,
    find_duplicate_rule,
)


def _session_returning(note):
    """A mocked async_session() whose single execute() yields `note` (or None)."""
    s = AsyncMock()
    s.__aenter__ = AsyncMock(return_value=s)
    s.__aexit__ = AsyncMock(return_value=False)
    result = MagicMock()
    result.scalars.return_value.first.return_value = note
    s.execute = AsyncMock(return_value=result)
    return s


def _fake_note(id=1, title="T", note_type="note"):
    n = MagicMock()
    n.id, n.title, n.note_type = id, title, note_type
    return n


@pytest.mark.asyncio
async def test_title_exact_match_returns_title_duplicate():
    note = _fake_note(id=10, title="Setup CI")
    with patch("scribe.services.dedup.async_session",
               return_value=_session_returning(note)):
        # whitespace/case differences are normalized away
        dup = await find_duplicate_note(7, "  setup   ci ", project_id=2, is_task=True)
    assert dup is not None
    assert dup.id == 10
    assert dup.reason == "title"
    assert dup.similarity == 1.0


@pytest.mark.asyncio
async def test_short_body_skips_semantic_check():
    sem = AsyncMock()
    with patch("scribe.services.dedup.async_session",
               return_value=_session_returning(None)), \
         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
        dup = await find_duplicate_note(7, "Unique", body="too short", project_id=2)
    assert dup is None
    sem.assert_not_called()  # body under _MIN_BODY_FOR_SEMANTIC


@pytest.mark.asyncio
async def test_semantic_match_when_body_substantial():
    hit = _fake_note(id=20, title="Existing", note_type="note")
    sem = AsyncMock(return_value=[(0.93, hit)])
    with patch("scribe.services.dedup.async_session",
               return_value=_session_returning(None)), \
         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
        dup = await find_duplicate_note(
            7, "Title", body="x" * 250, project_id=2, is_task=False, note_type="note",
        )
    assert dup is not None
    assert dup.id == 20
    assert dup.reason == "semantic"
    assert dup.similarity == 0.93


@pytest.mark.asyncio
async def test_semantic_match_of_other_note_type_is_ignored():
    other = _fake_note(id=21, title="X", note_type="process")
    sem = AsyncMock(return_value=[(0.97, other)])
    with patch("scribe.services.dedup.async_session",
               return_value=_session_returning(None)), \
         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
        dup = await find_duplicate_note(7, "Title", body="x" * 250, note_type="note")
    assert dup is None  # type mismatch must not block


@pytest.mark.asyncio
async def test_rule_title_match_in_topic():
    rule = _fake_note(id=47, title="Honor the multi-user sharing ACL")
    with patch("scribe.services.dedup.async_session",
               return_value=_session_returning(rule)):
        dup = await find_duplicate_rule(
            "honor the   multi-user sharing acl", topic_id=7,
        )
    assert dup is not None
    assert dup.id == 47
    assert dup.reason == "title"


@pytest.mark.asyncio
async def test_rule_requires_a_scope():
    # No topic_id and no project_id → nothing to scope to → no match, no query.
    sess = AsyncMock()
    with patch("scribe.services.dedup.async_session", return_value=sess):
        dup = await find_duplicate_rule("anything")
    assert dup is None
    sess.__aenter__.assert_not_called()


def test_duplicate_response_shape():
    dm = DuplicateMatch(id=5, title="Foo", similarity=1.0, reason="title")
    r = duplicate_response(dm, "task")
    assert r["duplicate"] is True
    assert r["existing_id"] == 5
    assert r["match"] == "title"
    assert "force=true" in r["message"]
    assert "update_task" in r["message"]


# --- snippet structural identity (#2518) -------------------------------------
#
# The gate used to compare a snippet's rendered DOCUMENT, which is mostly prose
# about the code. Measured on the button corpus, that failed in both directions
# at once: two deliberately-parallel variants were refused at 0.92, while a
# verbatim re-record of one snippet under a different name scored below 0.90 and
# was created. These tests pin the structural signals that replaced it.


def _session_sequence(results):
    """A mocked async_session() whose successive execute() calls yield `results`.

    The single-result helper above can't express this: the structural check runs
    a location query and then a code query, and the whole point is that they
    answer differently.
    """
    s = AsyncMock()
    s.__aenter__ = AsyncMock(return_value=s)
    s.__aexit__ = AsyncMock(return_value=False)
    wrapped = []
    for note in results:
        r = MagicMock()
        r.scalars.return_value.first.return_value = note
        wrapped.append(r)
    s.execute = AsyncMock(side_effect=wrapped)
    return s


@pytest.mark.asyncio
async def test_same_location_is_a_duplicate_however_it_is_described():
    """The measured false NEGATIVE: identical code at an identical
    repo·path·symbol was created because the prose around it differed."""
    existing = _fake_note(id=30, title=".btn-primary — a page's main action",
                          note_type="snippet")
    sem = AsyncMock()
    with patch("scribe.services.dedup.async_session",
               return_value=_session_sequence([None, existing])), \
         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
        dup = await find_duplicate_note(
            7, "primaryButton — something else entirely", body="x" * 400,
            project_id=2, is_task=False, note_type="snippet",
            code=".btn-primary { color: red; }",
            locations=[{"repo": "Scribe", "path": "a/b.css", "symbol": ".btn-primary"}],
        )
    assert dup is not None
    assert dup.reason == "location"
    assert dup.similarity == 1.0
    # Structural identity is certain, so it must not be diluted by asking the
    # embedder for a second opinion.
    sem.assert_not_called()


@pytest.mark.asyncio
async def test_identical_code_is_a_duplicate_at_a_different_location():
    existing = _fake_note(id=31, title="group_pairs", note_type="snippet")
    with patch("scribe.services.dedup.async_session",
               return_value=_session_sequence([None, None, existing])), \
         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes",
               AsyncMock(return_value=[])):
        dup = await find_duplicate_note(
            7, "unionFind", body="x" * 400, project_id=2, is_task=False,
            note_type="snippet", code="def f():\n    return 1",
            locations=[{"repo": "Scribe", "path": "z.py", "symbol": "f"}],
        )
    assert dup is not None
    assert dup.reason == "code"


@pytest.mark.asyncio
async def test_a_location_without_a_symbol_is_not_an_identity():
    """A path alone is a DIRECTORY of artefacts. Matching on it would refuse
    every second snippet recorded from one file — which is exactly the corpus
    the button recipes form."""
    session = _session_sequence([None])
    sem = AsyncMock(return_value=[])
    with patch("scribe.services.dedup.async_session", return_value=session), \
         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
        dup = await find_duplicate_note(
            7, "Some recipe", body="x" * 400, project_id=2, is_task=False,
            note_type="snippet", code="",
            locations=[{"repo": "Scribe", "path": "a/b.css", "symbol": ""}],
        )
    assert dup is None
    # Exactly one query — the title check. With no symbol and no code there is
    # nothing to match structurally, and the sequence above would raise
    # StopIteration if a second query were issued.
    assert session.execute.await_count == 1


@pytest.mark.asyncio
async def test_snippets_use_the_raised_semantic_threshold():
    """Variants of one component legitimately reach 0.92. The semantic arm has
    to sit above that band or it refuses the corpus it exists to protect."""
    from scribe.services.dedup import (
        _SEMANTIC_THRESHOLD,
        _SNIPPET_SEMANTIC_THRESHOLD,
    )
    sem = AsyncMock(return_value=[])
    with patch("scribe.services.dedup.async_session",
               return_value=_session_sequence([None, None, None])), \
         patch("scribe.services.dedup.embeddings_svc.semantic_search_notes", sem):
        await find_duplicate_note(
            7, "A recipe", body="x" * 400, project_id=2, is_task=False,
            note_type="snippet", code="x",
            locations=[{"repo": "R", "path": "p", "symbol": "s"}],
        )
    assert sem.await_args.kwargs["threshold"] == _SNIPPET_SEMANTIC_THRESHOLD
    assert _SNIPPET_SEMANTIC_THRESHOLD > 0.92, (
        "the observed sibling band tops out at 0.92 (.btn-danger vs "
        ".btn-danger-outline); a threshold at or below it blocks legitimate "
        "variants again"
    )
    assert _SNIPPET_SEMANTIC_THRESHOLD > _SEMANTIC_THRESHOLD


def test_sibling_variants_are_not_reported_as_merge_candidates():
    """The measured false POSITIVE: eight button recipes, every direct pair over
    the floor, proposed as ONE merge set."""
    from scribe.services.dedup import _drop_sibling_pairs

    records = {
        1: {"locations": [{"repo": "S", "path": "c.css", "symbol": ".btn-primary"}],
            "code_sha": "aaa"},
        2: {"locations": [{"repo": "S", "path": "c.css", "symbol": ".btn-secondary"}],
            "code_sha": "bbb"},
    }
    assert _drop_sibling_pairs([(1, 2, 0.87)], records) == []


def test_identical_code_still_reports_even_with_different_symbols():
    """The filter keys on "the author named these apart", but a shared code
    fingerprint overrides that — the same code under two names IS the
    copy-paste the report exists to surface."""
    from scribe.services.dedup import _drop_sibling_pairs

    records = {
        1: {"locations": [{"repo": "S", "path": "a.py", "symbol": "debounce"}],
            "code_sha": "same"},
        2: {"locations": [{"repo": "S", "path": "b.py", "symbol": "useDebounced"}],
            "code_sha": "same"},
    }
    assert _drop_sibling_pairs([(1, 2, 0.9)], records) == [(1, 2, 0.9)]


def test_unnamed_snippets_still_report():
    """A snippet with no recorded symbol made no identity claim, so the filter
    must not protect it — re-recording without a location is a common way to
    duplicate."""
    from scribe.services.dedup import _drop_sibling_pairs

    records = {1: {"code_sha": "aaa"}, 2: {"code_sha": "bbb"}}
    assert _drop_sibling_pairs([(1, 2, 0.9)], records) == [(1, 2, 0.9)]


def test_duplicate_response_names_what_matched_for_structural_hits():
    """A structural hit is certain, so the message must not hedge with
    "similar" — and it points at merge, which is what two records of one
    artefact actually need."""
    r = duplicate_response(
        DuplicateMatch(id=9, title=".btn-primary", similarity=1.0, reason="location"),
        "snippet",
    )
    assert "repo · path · symbol" in r["message"]
    assert "merge_snippets" in r["message"]
    assert "similar" not in r["message"]

    r = duplicate_response(
        DuplicateMatch(id=9, title="x", similarity=1.0, reason="code"), "snippet",
    )
    assert "identical code" in r["message"]