FabledScribe/tests/test_note_usage.py

"""Tests for the usage signal (#2085) — was a surfaced record ever pulled?

Covers the payload shaping and the two contracts that make this safe to leave in
the hot path: telemetry never raises, and telemetry never blocks. Plus the thing
the feature exists for — that the write-path PLACE arm is now recorded, since
before this it surfaced snippets while leaving no trace anywhere.
"""
from unittest.mock import AsyncMock, MagicMock, patch

import pytest


pytestmark = pytest.mark.usefixtures("_no_supersession")


from scribe.services import note_usage
from scribe.services.note_usage import (
    empty_usage,
    record_pulled,
    record_surfaced,
    usage_for_notes,
)
from tests.helpers import fake_note, writepath_cfg


# --- recording ------------------------------------------------------------


async def test_record_surfaced_writes_one_row_per_note():
    with patch.object(note_usage, "_schedule") as sched:
        record_surfaced(user_id=7, note_ids=[11, 12], source="auto_inject")
    rows = sched.call_args[0][0]
    assert [r["note_id"] for r in rows] == [11, 12]
    assert {r["event"] for r in rows} == {"surfaced"}
    assert {r["source"] for r in rows} == {"auto_inject"}
    assert {r["user_id"] for r in rows} == {7}


async def test_record_pulled_writes_a_single_row():
    with patch.object(note_usage, "_schedule") as sched:
        record_pulled(user_id=7, note_id=11, source="mcp_get_snippet")
    rows = sched.call_args[0][0]
    assert rows == [
        {"user_id": 7, "note_id": 11, "event": "pulled",
         "source": "mcp_get_snippet", "project_id": None}
    ]


async def test_empty_menu_schedules_nothing():
    """No notes surfaced is not an event — it must not cost a write."""
    with patch.object(note_usage, "_insert_events") as ins:
        record_surfaced(user_id=1, note_ids=[], source="auto_inject")
    ins.assert_not_called()


async def test_recording_never_raises_on_bad_input():
    """Telemetry sits in the hot path of every retrieval. A malformed id must
    cost a data point, never the operator's request."""
    with patch.object(note_usage, "_schedule"):
        record_surfaced(user_id=1, note_ids=["not-an-int"], source="auto_inject")
        record_pulled(user_id=1, note_id=None, source="mcp_get_note")


async def test_recording_without_an_event_loop_is_skipped_not_raised():
    """Called from a sync context outside the app (a script, a test helper),
    there is no loop to schedule on. Skip rather than blow up."""
    with patch.object(
        note_usage.asyncio, "get_running_loop", side_effect=RuntimeError
    ):
        record_pulled(user_id=1, note_id=5, source="mcp_get_note")


# --- readout --------------------------------------------------------------


async def test_usage_for_notes_zero_fills_every_requested_id():
    """The caller renders this shape unconditionally, so a note with no events
    must come back as zeroes, not as a missing key."""
    session = MagicMock()
    session.execute = AsyncMock(return_value=MagicMock(all=MagicMock(return_value=[])))
    ctx = MagicMock()
    ctx.__aenter__ = AsyncMock(return_value=session)
    ctx.__aexit__ = AsyncMock(return_value=False)
    with patch.object(note_usage, "async_session", return_value=ctx):
        out = await usage_for_notes([3, 4])
    assert out == {3: empty_usage(), 4: empty_usage()}


async def test_usage_for_notes_splits_counts_by_event():
    from datetime import datetime, timezone

    ts = datetime(2026, 7, 28, tzinfo=timezone.utc)
    # Rows are (note_id, event, count, last_at, ambient) since #2477 split the
    # readout. Ranked and ambient surfacings arrive as separate groups.
    rows = [
        (3, "surfaced", 9, ts, False),
        (3, "surfaced", 40, ts, True),
        (3, "pulled", 2, ts, False),
    ]
    session = MagicMock()
    # Two aggregates in one session: every event, then the away-from-home
    # subset (note_id, event, count) — none here.
    session.execute = AsyncMock(side_effect=[
        MagicMock(all=MagicMock(return_value=rows)),
        MagicMock(all=MagicMock(return_value=[])),
    ])
    ctx = MagicMock()
    ctx.__aenter__ = AsyncMock(return_value=session)
    ctx.__aexit__ = AsyncMock(return_value=False)
    with patch.object(note_usage, "async_session", return_value=ctx):
        out = await usage_for_notes([3])
    # The dead-weight reading ("surfaced often, never pulled") is only valid
    # over surfacings that were CHOICES. 40 enter_project appearances must not
    # make a record look popular — they sit in ambient_count (#2477).
    assert out[3]["surfaced_count"] == 9
    assert out[3]["ambient_count"] == 40
    assert out[3]["pull_count"] == 2
    assert out[3]["last_pulled_at"] == ts.isoformat()


async def test_usage_for_notes_counts_what_happened_away_from_home():
    """Surfaced or opened on a project other than the record's own is the
    evidence a record transferred (#3735). It is a SUBSET of the totals, read
    from the second aggregate, and it must land on the right counter."""
    from datetime import datetime, timezone

    ts = datetime(2026, 10, 1, tzinfo=timezone.utc)
    session = MagicMock()
    session.execute = AsyncMock(side_effect=[
        MagicMock(all=MagicMock(return_value=[
            (3, "surfaced", 9, ts, False),
            (3, "pulled", 4, ts, False),
            (4, "surfaced", 2, ts, False),
        ])),
        MagicMock(all=MagicMock(return_value=[
            (3, "surfaced", 5),
            (3, "pulled", 1),
        ])),
    ])
    ctx = MagicMock()
    ctx.__aenter__ = AsyncMock(return_value=session)
    ctx.__aexit__ = AsyncMock(return_value=False)
    with patch.object(note_usage, "async_session", return_value=ctx):
        out = await usage_for_notes([3, 4])
    assert (out[3]["surfaced_away_count"], out[3]["pulled_away_count"]) == (5, 1)
    assert (out[3]["surfaced_count"], out[3]["pull_count"]) == (9, 4)
    # Used only at home: the away counters stay zero, not missing.
    assert (out[4]["surfaced_away_count"], out[4]["pulled_away_count"]) == (0, 0)


async def test_usage_readout_failure_degrades_to_zeroes():
    """A telemetry readout must not be able to break the list it decorates."""
    with patch.object(note_usage, "async_session", side_effect=RuntimeError("boom")):
        out = await usage_for_notes([3])
    assert out == {3: empty_usage()}


async def test_no_ids_short_circuits_without_a_query():
    with patch.object(note_usage, "async_session") as sess:
        assert await usage_for_notes([]) == {}
    sess.assert_not_called()


# --- the gap this closes --------------------------------------------------


@pytest.mark.parametrize(
    "lookups, expected_source",
    [
        # A hit AT the exact file is the sync class (#2708); a hit from the
        # directory query is the reuse-shaped place arm. Both are un-scored,
        # and both must leave a usage trace under their own name.
        ([(1, "hit"), (2, "empty")], "write_path_sync"),
        ([(1, "empty"), (2, "hit")], "write_path_place"),
    ],
    ids=["at-the-file", "nearby"],
)
async def test_unscored_location_arms_are_recorded(lookups, expected_source):
    """The location arms carry no score, so they have no home in retrieval_logs
    — they surfaced snippets while leaving no trace anywhere. That was the
    blocker #2082 recorded against this task; this is the assertion that it's
    closed, per class."""
    from scribe.services import plugin_context

    here = [{"id": 42, "title": "helper", "user_id": 1, "note_type": "snippet"}]
    responses = [(here, 1) if kind == "hit" else ([], 0) for _n, kind in lookups]
    with (
        patch.object(
            plugin_context,
            "get_writepath_config",
            AsyncMock(return_value=writepath_cfg(threshold=0.55)),
        ),
        patch.object(
            plugin_context.snippets_svc,
            "list_snippets",
            AsyncMock(side_effect=responses),
        ),
        patch.object(
            plugin_context, "semantic_search_notes", AsyncMock(return_value=[])
        ),
        patch.object(plugin_context, "owner_names_for", AsyncMock(return_value={})),
        patch.object(plugin_context, "record_retrieval"),
        patch.object(plugin_context, "record_surfaced") as surfaced,
    ):
        out = await plugin_context.build_write_path_hint(
            1, "src/a.py", code="def f(): pass"
        )

    assert out["note_ids"] == [42]
    sources = {c.kwargs["source"] for c in surfaced.call_args_list}
    assert expected_source in sources


async def test_auto_inject_records_what_survived_the_margin_gate():
    """Not what the ranker returned — retrieval_logs already holds that. These
    two numbers must not silently mean different things per surface."""
    from scribe.services import plugin_context

    hits = [(0.90, fake_note(id=1, title="kept", user_id=1, note_type="snippet")), (0.40, fake_note(id=2, title="cut by the margin gate", user_id=1, note_type="snippet"))]
    with (
        patch.object(
            plugin_context,
            "get_autoinject_config",
            AsyncMock(return_value={"enabled": True, "threshold": 0.3, "top_k": 5}),
        ),
        patch.object(
            plugin_context, "semantic_search_notes", AsyncMock(return_value=hits)
        ),
        patch.object(plugin_context, "owner_names_for", AsyncMock(return_value={})),
        patch.object(plugin_context, "record_retrieval"),
        patch.object(plugin_context, "record_surfaced") as surfaced,
    ):
        await plugin_context.build_autoinject_hint(1, "a query")

    assert surfaced.call_args.kwargs["note_ids"] == [1]
    assert surfaced.call_args.kwargs["source"] == "auto_inject"


def test_every_getter_that_can_be_surfaced_also_records_a_pull():
    """Rule #33 contract check — and the one that would have caught #2245.

    `surfaced` and `pulled` only mean something as a PAIR: the rate between them
    is what #1038 and #2085 gate on. That pair is only closed if the tool which
    OPENS a record reports it. `get_note` did, `get_snippet` did, `get_task` did
    NOT — and auto-inject ranks kind-blind over a corpus that is overwhelmingly
    tasks and issues, so the gap sat exactly where the volume is: every surfaced
    task counted as never-pulled, dragging measured pull-through toward zero for
    the menu's own dominant kind.

    Asserted by source inspection rather than by calling the tools, because the
    failure is a MISSING call — which no behavioural test of the tool's return
    value can see.
    """
    import inspect

    from scribe.mcp.tools import notes as notes_tools
    from scribe.mcp.tools import snippets as snippet_tools
    from scribe.mcp.tools import tasks as task_tools

    getters = (
        (notes_tools, "get_note"),
        (task_tools, "get_task"),
        (snippet_tools, "get_snippet"),
    )
    for module, name in getters:
        src = inspect.getsource(getattr(module, name))
        assert "record_pulled(" in src, (
            f"{name} can be surfaced in an auto-inject menu but records no pull — "
            "its pull-through rate will read as zero regardless of real usage"
        )


# --- persistence (integration) --------------------------------------------
# Everything above mocks _schedule or the session — deliberately, for the hot
# path. But that left the two functions that actually touch the database
# (_insert_events and usage_for_notes' real SQL) running against real Postgres
# nowhere, which is how the deployed instance reported zero for every counter
# while surfacing demonstrably fired (#2663): all-green mocked units over a
# dead real path, the #2109 shape. These two run in the CI integration lane
# and split the chain so a failure names its half.


async def _purge(note_id: int) -> None:
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.note_usage import NoteUsageEvent

    async with async_session() as s:
        await s.execute(
            delete(NoteUsageEvent).where(NoteUsageEvent.note_id == note_id)
        )
        await s.commit()


@pytest.mark.integration
async def test_insert_and_readout_roundtrip_on_real_postgres(_dispose_engine):
    """WRITE half + READ half against the real table, one assertion per counter."""
    from scribe.services.note_usage import _insert_events

    nid = 990101
    try:
        await _insert_events([
            {"user_id": 7, "note_id": nid, "event": "surfaced",
             "source": "write_path_place"},
            {"user_id": 7, "note_id": nid, "event": "surfaced",
             "source": "enter_project"},
            {"user_id": 7, "note_id": nid, "event": "pulled",
             "source": "mcp_get_snippet"},
        ])
        out = await usage_for_notes([nid])
        # write_path_place is a ranked choice; enter_project is ambient (#2477).
        assert out[nid]["surfaced_count"] == 1
        assert out[nid]["ambient_count"] == 1
        assert out[nid]["pull_count"] == 1
        assert out[nid]["last_surfaced_at"] is not None
        assert out[nid]["last_pulled_at"] is not None
    finally:
        await _purge(nid)


@pytest.mark.integration
async def test_record_pulled_lands_end_to_end_from_a_running_loop(_dispose_engine):
    """The exact chain the deployed instance runs: record_pulled schedules a
    fire-and-forget task on the running loop, and the row must land. The
    _pending set (which exists to keep the loop's weak-ref'd tasks alive) is
    also what lets this test await a write that is fire-and-forget by design."""
    import asyncio

    nid = 990102
    try:
        record_pulled(user_id=7, note_id=nid, source="mcp_get_snippet")
        assert note_usage._pending, "record_pulled scheduled no task"
        await asyncio.gather(*note_usage._pending)
        out = await usage_for_notes([nid])
        assert out[nid]["pull_count"] == 1
    finally:
        await _purge(nid)


@pytest.mark.integration
async def test_away_counts_compare_the_readers_project_with_the_records(_dispose_engine):
    """The real join: an event counts as away only when the reader's project is
    known and differs from the record's own. Home, unreported and ambient
    events are all left out — each would otherwise read as transfer."""
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.note import Note
    from scribe.models.project import Project
    from scribe.services.note_usage import _insert_events
    from tests.helpers import ensure_user

    async with async_session() as s:
        owner = await ensure_user(s, "usage_away_owner")
        home = Project(user_id=owner.id, title="usage home")
        away = Project(user_id=owner.id, title="usage away")
        s.add_all([home, away])
        await s.flush()
        lesson = Note(user_id=owner.id, project_id=home.id, title="a lesson",
                      note_type="lesson")
        s.add(lesson)
        await s.flush()
        nid, home_id, away_id, uid = lesson.id, home.id, away.id, owner.id
        await s.commit()
    try:
        def ev(event, source, pid):
            return {"user_id": uid, "note_id": nid, "event": event,
                    "source": source, "project_id": pid}
        await _insert_events([
            ev("surfaced", "lesson_slot", away_id),     # away: counts
            ev("surfaced", "lesson_slot", home_id),     # home
            ev("surfaced", "lesson_slot", None),        # unreported
            ev("surfaced", "enter_project", away_id),   # ambient
            ev("pulled", "mcp_get_lesson", away_id),    # away: counts
            ev("pulled", "mcp_get_lesson", None),       # unreported
        ])
        out = await usage_for_notes([nid])
        assert out[nid]["surfaced_away_count"] == 1
        assert out[nid]["pulled_away_count"] == 1
        assert out[nid]["surfaced_count"] == 3
        assert out[nid]["pull_count"] == 2
    finally:
        await _purge(nid)
        async with async_session() as s:
            await s.execute(delete(Note).where(Note.id == nid))
            await s.execute(delete(Project).where(Project.id.in_([home_id, away_id])))
            await s.commit()