FabledGrimoire/tests/test_ledger_references_not_silence.py

"""A repeat on a note arm is referenced, not withheld.

WHY THIS EXISTS

A record the session was told about an hour ago is not a record in front of the reader now, so the second time it is
the best answer the session gets the line again with a tail saying so — never
silence, which is indistinguishable from "nothing matched".

The note and snippet arms hold the same line. A ledger passed into
`semantic_search_notes` as `exclude_ids` takes the repeat out of the candidate
set entirely, with three consequences:

  - the session gets silence the second time, on the arms that fire most;
  - a compaction makes it permanent, because the ledger outlives the context it
    describes;
  - and `best_available` is measured against a candidate set the caller has
    already edited, so the bar can be blamed for a record the caller withheld.

WHAT THIS PINS

  1. The ledger does not reach the search. Asserted on the call's kwargs,
     because this is the difference between a reference and silence and every
     behavioural assertion below rests on it.
  2. The repeat is rendered, and marked. The wording is NOT the rule arms' —
     "before deciding it does not apply" is the voice of a record that binds,
     and a dev-log borrowing it would claim authority it does not have — so
     what is pinned is that the line appears and is distinguishable, not its
     prose.
  3. The telemetry splits the two: the row's result set and the
     surfaced table both take FRESH only, and the repeat is COUNTED in
     `suppressed` instead. This is what makes a zero-result call readable —
     `result_count == 0` with `suppressed_count > 0` says "everything that
     matched, the session has already seen".
  4. The deliberate exception: the write-path SYNC class still shows once. Its
     claim is about an edit in progress, not about a record's continuing
     relevance, and repeating it would be nagging rather than reminding.
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from tests.helpers import fake_note, writepath_cfg

REAL_CODE = '''def debounce(fn, wait=0.25):
    """Rate-limit a callback so it fires once after the last call."""
    timer = None

    def wrapped(*a, **kw):
        nonlocal timer
        if timer:
            timer.cancel()
        timer = threading.Timer(wait, fn, a, kw)
        timer.start()

    return wrapped
'''


def _wp_cfg(**over):
    return writepath_cfg(**over)


def _snippet_item(nid, title, user_id=1):
    return {"id": nid, "title": title, "user_id": user_id, "note_type": "snippet"}


# ── auto-inject ────────────────────────────────────────────────────────────

async def _autoinject(hits, exclude_ids, *, rec=None, surf=None):
    from grimoire.services import plugin_context as pc
    search = AsyncMock(return_value=hits)
    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, "record_retrieval", rec or MagicMock()), \
         patch.object(pc, "record_surfaced", surf or MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
         patch.object(pc, "superseded_ids", AsyncMock(return_value=set())):
        out = await pc.build_autoinject_hint(
            1, "postgres pool", project_id=2, exclude_ids=exclude_ids)
    return out, search


@pytest.mark.asyncio
async def test_the_ledger_never_reaches_the_search():
    """The load-bearing one. A ledger inside the query IS the silence."""
    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1))]
    _out, search = await _autoinject(hits, [11])
    # The menu arm runs first; the reuse slot's own call may legitimately
    # exclude this call's menu, which is a different claim (see below).
    menu_call = search.call_args_list[0]
    assert not menu_call.kwargs.get("exclude_ids"), (
        "the session ledger was passed into the search, so a repeat is "
        "withheld rather than referenced"
    )


@pytest.mark.asyncio
async def test_a_repeat_is_shown_again_and_marked():
    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
            (0.78, fake_note(id=22, title="run_maintenance thresholds", user_id=1))]
    out, _ = await _autoinject(hits, [11])
    assert "#11" in out["context"] and "#22" in out["context"]
    # Distinguishable, without pinning the word's neighbours in the sentence.
    line_11 = next(ln for ln in out["context"].splitlines() if "#11" in ln)
    line_22 = next(ln for ln in out["context"].splitlines() if "#22" in ln)
    assert "seen" in line_11 and "seen" not in line_22


@pytest.mark.asyncio
async def test_the_repeat_is_counted_not_reported_as_a_result():
    rec = MagicMock()
    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
            (0.78, fake_note(id=22, title="run_maintenance thresholds", user_id=1))]
    await _autoinject(hits, [11], rec=rec)
    row = next(c.kwargs for c in rec.call_args_list
               if c.kwargs["source"] == "auto_inject")
    assert [int(n.id) for _s, n in row["results"]] == [22]
    assert row["suppressed"] == 1


@pytest.mark.asyncio
async def test_a_call_where_everything_was_already_seen_is_readable():
    """A call where every hit was already seen is not a call nothing cleared.

    Logged as zero results with no suppression count, it would be
    indistinguishable from a bar nothing cleared — and the operator tuning
    that bar would be reading the wrong number.
    """
    rec = MagicMock()
    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1))]
    out, _ = await _autoinject(hits, [11], rec=rec)
    row = next(c.kwargs for c in rec.call_args_list
               if c.kwargs["source"] == "auto_inject")
    assert row["results"] == [] and row["suppressed"] == 1
    # And the session still gets the line, which is the whole point.
    assert "#11" in out["context"]


@pytest.mark.asyncio
async def test_a_repeat_is_not_recorded_as_a_fresh_surfacing():
    """The two-table identity: this table and the log row describe the same call."""
    surf = MagicMock()
    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1)),
            (0.78, fake_note(id=22, title="run_maintenance thresholds", user_id=1))]
    await _autoinject(hits, [11], surf=surf)
    rows = [c.kwargs for c in surf.call_args_list
            if c.kwargs.get("source") == "auto_inject"]
    assert rows and rows[0]["note_ids"] == [22]


@pytest.mark.asyncio
async def test_the_header_explains_the_seen_marker():
    """A contract stated in the prose is a contract: the header explains the
    `seen` marker and does not promise a record once per session.

    Cheap to forget and invisible when wrong: the menu would carry a marker
    the header had never explained, and a reader would have to guess.
    """
    hits = [(0.80, fake_note(id=11, title="Pool sizing decision", user_id=1))]
    out, _ = await _autoinject(hits, [])
    header = out["context"].splitlines()[0]
    assert "once per session" not in header
    assert "seen" in header


# ── the write path ─────────────────────────────────────────────────────────

async def _write_path(hits, exclude_ids, *, here=(), rec=None, sync_exclude=()):
    from grimoire.services import plugin_context as pc
    search = AsyncMock(return_value=hits)
    listing = AsyncMock(side_effect=[(list(here), len(here)), ([], 0)])
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_wp_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets", listing), \
         patch.object(pc, "semantic_search_notes", search), \
         patch.object(pc, "record_retrieval", rec or MagicMock()), \
         patch.object(pc, "record_surfaced", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(
            1, "src/x.py", code=REAL_CODE,
            exclude_ids=list(exclude_ids),
            exclude_sync_ids=list(sync_exclude),
        )
    return out, search


@pytest.mark.asyncio
async def test_the_write_path_ledger_does_not_reach_its_search_either():
    hits = [(0.83, fake_note(id=11, title="debounce helper", user_id=1))]
    _out, search = await _write_path(hits, [11])
    assert 11 not in (search.call_args.kwargs.get("exclude_ids") or set())


@pytest.mark.asyncio
async def test_a_write_path_repeat_is_rendered_with_a_marker():
    hits = [(0.83, fake_note(id=11, title="debounce helper", user_id=1))]
    out, _ = await _write_path(hits, [11])
    assert "#11" in out["context"]
    assert "seen" in next(ln for ln in out["context"].splitlines() if "#11" in ln)


@pytest.mark.asyncio
async def test_the_write_path_counts_its_repeat():
    rec = MagicMock()
    hits = [(0.83, fake_note(id=11, title="debounce helper", user_id=1)),
            (0.80, fake_note(id=22, title="throttle helper", user_id=1))]
    await _write_path(hits, [11], rec=rec)
    row = next(c.kwargs for c in rec.call_args_list
               if c.kwargs["source"] == "write_path")
    assert [int(n.id) for _s, n in row["results"]] == [22]
    assert row["suppressed"] == 1


@pytest.mark.asyncio
async def test_this_calls_own_menu_is_still_excluded_from_its_search():
    """The claim that stayed an exclusion, and must not be lost with the other.

    A snippet already listed by PLACE in this same hint has nothing to gain
    from a second line in it. That is same-call duplication, not a repeat
    across calls, and the two were the same variable until this step.
    """
    hits = []
    _out, search = await _write_path(
        hits, [], here=[_snippet_item(7, "records this file")])
    assert 7 in (search.call_args.kwargs.get("exclude_ids") or set())


@pytest.mark.asyncio
async def test_the_sync_class_still_shows_only_once():
    """The deliberate exception, pinned so it reads as a decision.

    The sync nudge says "you are editing the file this record describes, so
    updating it is part of the edit". Repeated every write to the same file it
    is nagging, and unlike a reuse suggestion it is not a claim whose relevance
    can return — it either got acted on or it did not.
    """
    out, _ = await _write_path(
        [], [], here=[_snippet_item(7, "records this file")], sync_exclude=[7])
    assert out["sync_note_ids"] == []
    assert "#7" not in out["context"]