FabledGrimoire/tests/test_named_records.py

"""A record the operator names by number is looked up, not left to ranking.

"yes go ahead with 4448" says exactly which record is meant, and a menu
built by ranking cannot use it: a number means nothing to an embedding, so the
menu fills with whatever resembles the words around it. The parser cases below
are drawn from real operator prompts, and the half that must NOT be read as a
reference matters as much as the half that must: every misread number is a
stranger's record dropped in front of the reader.
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from grimoire.services.record_refs import NAMED_LIMIT, named_record_ids
from tests.helpers import fake_note


# ── The parser ─────────────────────────────────────────────────────────────

NAMES = [
    ("yes go ahead with 4448", [4448]),
    ("4364 - we had talked about the follow-up case", [4364]),
    ("4417 sounds like the right call", [4417]),
    ("awesome go for 3898", [3898]),
    ("let's continue #4772 then", [4772]),
    ("does #12 still hold?", [12]),
    ("look at 4448, 4449 and 4450", [4448, 4449, 4450]),
    ("with #4448 & #4449", [4448, 4449]),
    ("merged (#4794) this morning", [4794]),
    ("task 4448 first, then #4449", [4448, 4449]),
    ("#4448 again, and #4448", [4448]),               # once, in mention order
]

NOT_NAMES = [
    "merge PR #198 to main",
    "PR #198 and #199 are both green",                # a list inherits its head
    "CI #7855 went red",
    "rule 153 says plain merges",
    "rule #153 says plain merges",
    "milestone 444 is next",
    "wait for 300 seconds first",
    "about 500 records came back",
    "move the budget 3 to 6",
    "it ran 7855 times",                              # no reference word before it
    "bump it to 1.2.300",
    "on 2026-10-03 it broke",
    "see https://forge/x/pulls/198",
    "the cut is at 1,000",
    "costs $250 a month",
    "use 300ms",
    "rank #1 was unrelated",                          # one digit after `#`
    "with 42 of them",                                # two digits, bare
    "```\nfor 4448 in range\n```",                    # fenced code names nothing
    "",
]


@pytest.mark.parametrize("prompt,want", NAMES, ids=[p for p, _ in NAMES])
def test_a_named_record_is_read(prompt, want):
    assert named_record_ids(prompt) == want


@pytest.mark.parametrize("prompt", NOT_NAMES, ids=[p[:30] or "empty" for p in NOT_NAMES])
def test_a_number_that_is_not_a_reference_is_left(prompt):
    assert named_record_ids(prompt) == []


def test_a_long_list_is_capped():
    prompt = "with " + ", ".join(str(1000 + i) for i in range(NAMED_LIMIT + 3))
    assert len(named_record_ids(prompt)) == NAMED_LIMIT


# ── The menu ───────────────────────────────────────────────────────────────


async def _hint(prompt, *, hits=(), records=None, seen=(), context=""):
    """Run the prompt arm with the search, lookup and recorders stubbed.

    `records` is what the ACL lookup can see, by id; anything else is
    inaccessible.
    """
    from grimoire.services import plugin_context as pc

    records = records or {}
    calls: list[int] = []

    async def _search(*_a, **_kw):
        calls.append(1)
        return list(hits) if len(calls) == 1 else []

    async def _get(_uid, nid):
        note = records.get(nid)
        return (note, "owner") if note is not None else None

    surfaced = MagicMock()
    retrieval = MagicMock()
    with patch.object(pc, "get_autoinject_config",
                      AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \
         patch.object(pc, "semantic_search_notes", _search), \
         patch.object(pc.notes_svc, "get_note_for_user", AsyncMock(side_effect=_get)), \
         patch.object(pc, "superseded_ids", AsyncMock(return_value=set())), \
         patch.object(pc, "system_names_for", AsyncMock(return_value={})), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
         patch.object(pc, "record_retrieval", retrieval), \
         patch.object(pc, "record_surfaced", surfaced):
        out = await pc.build_autoinject_hint(
            1, prompt, project_id=2, exclude_ids=list(seen), context=context,
        )
    by_source = {c.kwargs["source"]: c.kwargs["note_ids"] for c in surfaced.call_args_list}
    return out, by_source, retrieval


def _task(nid, title, body=""):
    return fake_note(id=nid, title=title, body=body, user_id=1,
                     is_task=True, status="todo", task_kind="issue")


@pytest.mark.asyncio
async def test_a_named_record_arrives_even_when_nothing_matched():
    """The case the whole change is for: a short prompt whose words match
    nothing still names its record, and the record arrives."""
    rec = _task(4448, "Filing runs itself after each scan",
                "## Symptom\nBooks sat unfiled until someone pressed the button.")
    out, surfaced, _ = await _hint("yes go ahead with 4448", records={4448: rec})

    ctx = out["context"]
    assert "Named in your message" in ctx
    assert '#4448 [issue (todo)] "Filing runs itself after each scan"' in ctx
    assert "↳ ## Symptom Books sat unfiled" in ctx
    assert "Possibly relevant" not in ctx, "a menu header over no menu lines"
    assert out["note_ids"] == [4448], "the named record must reach the session ledger"
    assert surfaced["named_ref"] == [4448]
    assert not surfaced.get("auto_inject"), "the ranking arm surfaced nothing here"


@pytest.mark.asyncio
async def test_a_named_record_comes_first_and_is_not_repeated_as_a_match():
    named = _task(4448, "the named one")
    other = fake_note(id=11, title="something that matched the words", user_id=1)
    hits = [(0.80, named), (0.78, other)]
    out, surfaced, retrieval = await _hint(
        "go ahead with 4448 and the filing fix", hits=hits, records={4448: named},
    )

    ctx = out["context"]
    assert ctx.index("Named in your message") < ctx.index("Possibly relevant")
    assert ctx.count("#4448") == 1, "the named record was shown twice"
    assert out["note_ids"] == [4448, 11]
    # Booked once, under the lookup — the ranking arm did not surface it, and
    # its log row counts it as suppressed so the row and the usage table agree.
    assert surfaced["named_ref"] == [4448]
    assert surfaced["auto_inject"] == [11]
    row = retrieval.call_args_list[0].kwargs
    assert row["source"] == "auto_inject"
    assert [int(n.id) for _s, n in row["results"]] == [11]
    assert row["suppressed"] == 1


@pytest.mark.asyncio
async def test_a_prompt_with_no_subject_still_shows_what_it_names():
    """"go ahead with 4448" names a record and nothing else: the record is
    looked up and shown, and no ranked menu is searched for."""
    rec = _task(4448, "the named one")
    other = fake_note(id=11, title="what the reply was about", user_id=1)
    out, surfaced, retrieval = await _hint(
        "go ahead with 4448", hits=[(0.8, other)], records={4448: rec},
        context="I'll start on the importer next.",
    )

    assert out["note_ids"] == [4448]
    assert "Possibly relevant" not in out["context"]
    assert "auto_inject" not in surfaced
    retrieval.assert_not_called()


@pytest.mark.asyncio
async def test_a_named_record_already_shown_is_a_pointer():
    rec = _task(4448, "the named one", "a long body that is already in context")
    out, surfaced, _ = await _hint("with 4448", records={4448: rec}, seen=[4448])

    assert "> - #4448 [issue (todo) · seen] the named one" in out["context"]
    assert "already in context" not in out["context"]
    assert "named_ref" not in surfaced, "a repeat is not a new surfacing"


@pytest.mark.asyncio
async def test_an_id_the_reader_cannot_open_is_dropped():
    """Not found, not theirs, or in the trash — all three say nothing, because
    the parser cannot tell a record id from a number it misread."""
    trashed = _task(4449, "gone")
    trashed.deleted_at = "2026-10-01T00:00:00Z"
    out, surfaced, _ = await _hint("with 4448 and 4449", records={4449: trashed})

    assert out["context"] == ""
    assert out["note_ids"] == []
    assert "named_ref" not in surfaced


@pytest.mark.asyncio
async def test_a_number_in_the_assistants_reply_is_not_named():
    """The query is enriched with the reply before a short prompt is searched;
    only the operator's own words can name a record."""
    rec = _task(4448, "the named one")
    out, _, _ = await _hint(
        "sounds good", records={4448: rec},
        context="I could start on #4448 next if you want.",
    )
    assert "Named in your message" not in out["context"]