FabledGrimoire/tests/test_autoinject_context.py

"""The auto-inject query carries the conversation beside a thin prompt.

The operator's typed words alone are a poor query: a mid-session follow-up —
"yes do that", "now fix the filter" — names nothing a record can match. The
hook reads the tail of the last assistant reply out
of the transcript and sends it as `ctx`; the server appends it to a SHORT
prompt only. Two halves, pinned separately: what the server builds, and what
the hook extracts.
"""
from __future__ import annotations

import json
import subprocess
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

from grimoire.services.plugin_context import (
    _AUTOINJECT_CONTEXT_MAX,
    _AUTOINJECT_CONTEXT_PROMPT_MAX,
    _autoinject_query,
    _has_subject,
    _menu_name,
    _menu_passage,
)
from tests.helpers import fake_note, need_tools

DEFS = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "grimoire_defs.sh"


# --- the query the server builds ---------------------------------------------

def test_a_short_prompt_is_followed_by_the_context():
    q = _autoinject_query("yes do that", "I'll move the library filters into the side column.")
    assert q.startswith("yes do that\n\n")
    assert "library filters" in q


def test_a_prompt_that_says_what_it_is_about_is_left_alone():
    long_prompt = "x" * (_AUTOINJECT_CONTEXT_PROMPT_MAX + 1)
    assert _autoinject_query(long_prompt, "anything at all") == long_prompt


def test_no_context_is_the_prompt_unchanged():
    assert _autoinject_query("yes do that", "") == "yes do that"
    assert _autoinject_query("yes do that", "   \n ") == "yes do that"


def test_context_is_cut_from_the_end_of_the_reply():
    # The reply's close is where it says what it did and what is next — the
    # part the operator's follow-up is answering.
    ctx = "HEAD " + "m" * 2000 + " TAIL"
    q = _autoinject_query("ok", ctx)
    tail = q.split("\n\n", 1)[1]
    assert len(tail) == _AUTOINJECT_CONTEXT_MAX
    assert tail.endswith("TAIL") and "HEAD" not in tail


# --- a prompt that names nothing ----------------------------------------------

@pytest.mark.parametrize("prompt", [
    "keep going", "yes please do", "1. yes 2. yes 3. yes", "start step 6",
    "is it deployed?", "awesome, what's next?", "go ahead",
    "is there anything else open in this project?",
    "please commit and let me know when it passes CI",
])
def test_a_turn_of_the_conversation_names_no_subject(prompt):
    assert not _has_subject(prompt)


@pytest.mark.parametrize("prompt", [
    "now fix the filter", "keep steward, it's our own app",
    "continue into DRY 14", "I've added a maintenance plan",
    "fix the Windows gate first, then step 2",
])
def test_one_word_of_its_own_is_a_subject(prompt):
    assert _has_subject(prompt)


@pytest.mark.asyncio
async def test_a_prompt_with_no_subject_gets_no_menu_and_no_search():
    """Its query would be the previous reply's tail alone, and the menu would
    hand back what that reply was built from."""
    from grimoire.services import plugin_context as pc

    search = AsyncMock(return_value=[(0.9, fake_note(id=11, title="the reply's own record", user_id=1))])
    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, "record_retrieval", retrieval), \
         patch.object(pc, "record_surfaced", MagicMock()):
        out = await pc.build_autoinject_hint(
            1, "yes please do", project_id=2,
            context="I'll move the library filters into the side column.",
        )

    assert out["context"] == "" and out["note_ids"] == []
    search.assert_not_called()
    retrieval.assert_not_called()


# --- what the hook extracts --------------------------------------------------

def _rec(role: str, blocks: list[dict], sidechain: bool = False) -> str:
    return json.dumps({
        "type": role, "isSidechain": sidechain,
        "message": {"role": role, "content": blocks},
    }, separators=(",", ":"))


def recent(tmp_path: Path, lines: list[str]) -> str:
    need_tools("bash", "awk", "grep", "tail")
    t = tmp_path / "t.jsonl"
    t.write_text("\n".join(lines) + "\n")
    r = subprocess.run(
        ["bash", "-c", f'set -uo pipefail\n. "{DEFS}"\ngrimoire_recent_context "{t}"'],
        capture_output=True,
    )
    assert r.returncode == 0, r.stderr.decode()
    return r.stdout.decode().strip()


def test_the_hook_reads_the_last_assistant_text(tmp_path):
    out = recent(tmp_path, [
        _rec("assistant", [{"type": "text", "text": "an older reply"}]),
        _rec("assistant", [{"type": "tool_use", "id": "t1", "name": "Bash", "input": {}}]),
        _rec("assistant", [{"type": "text", "text": "Shall I file it as an issue?\nNext: the hook."}]),
    ])
    assert out == "Shall I file it as an issue? Next: the hook."


def test_a_subagent_reply_is_not_this_session(tmp_path):
    out = recent(tmp_path, [
        _rec("assistant", [{"type": "text", "text": "the session's reply"}]),
        _rec("assistant", [{"type": "text", "text": "a subagent's report"}], sidechain=True),
    ])
    assert out == "the session's reply"


def test_the_hook_caps_what_it_sends(tmp_path):
    out = recent(tmp_path, [_rec("assistant", [{"type": "text", "text": "a" * 5000 + "END"}])])
    assert len(out) == 600 and out.endswith("END")


def test_no_transcript_is_silence_not_failure(tmp_path):
    assert recent(tmp_path, []) == ""


# --- what a menu line carries -----------------------------------------------
#
# The name, the kind and System, and the WHOLE matched passage — once. A repeat
# is a pointer. These pin the shape against the three ways it goes wrong:
# a trigger-composed title rendered as the line (1,500+ chars), the passage cut
# to its head (which was the title again), and a `seen` repeat re-rendered whole.

TRIGGER = "Adding a record type that is semantically searchable. " * 20


def test_a_snippet_line_shows_its_name_not_its_trigger():
    assert _menu_name(f"embed_x — {TRIGGER}", "snippet", {"name": "embed_x"}) == "embed_x"
    # With no mirror, the first separator is the seam (snippets.py's inverse).
    assert _menu_name(f"embed_x — {TRIGGER}", "snippet", None) == "embed_x"


def test_a_lesson_line_shows_its_subject_not_its_trigger():
    title = f"A guard does not undo a stored value — {TRIGGER.strip()}"
    name = _menu_name(title, "lesson", {"when_to_apply": TRIGGER.strip()})
    assert name == "A guard does not undo a stored value"


def test_a_plain_note_keeps_its_title_dashes_and_all():
    t = "Dev-log 2026-07-29 — milestone #232 closed"
    assert _menu_name(t, "note", None) == t


def test_the_passage_is_the_whole_chunk_without_its_title_prefix():
    body = "section " * 150  # ~1.2 KB: nothing of it is cut
    out = _menu_passage("Pool sizing", f"Pool sizing\n{body}\nsecond line")
    assert not out.startswith("Pool sizing")
    assert out.endswith("second line") and "\n" not in out
    assert len(out) > 1100


def test_a_title_only_match_hands_over_the_trigger_it_matched_on():
    title = "embed_x — when adding a searchable record"
    assert _menu_passage(title, title, "embed_x") == "when adding a searchable record"


async def _menu(hits, seen, chunks, systems=None):
    from grimoire.services import plugin_context as pc

    calls: list[int] = []

    async def _search(*_a, **kw):
        calls.append(1)
        if len(calls) > 1:
            return []
        if kw.get("report") is not None:
            kw["report"]["best_chunk"] = chunks
        return 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, "superseded_ids", AsyncMock(return_value=set())), \
         patch.object(pc, "system_names_for", AsyncMock(return_value=systems or {})), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "record_surfaced", MagicMock()):
        return (await pc.build_autoinject_hint(1, "q", project_id=2, exclude_ids=seen))["context"]


@pytest.mark.asyncio
async def test_a_first_sighting_carries_name_system_and_passage():
    title = f"embed_x — {TRIGGER}"
    hits = [(0.8, fake_note(id=11, title=title, note_type="snippet",
                            data={"name": "embed_x"}, user_id=1))]
    out = await _menu(hits, [], {11: {"index": 1, "text": f"{title}\nthe matched section"}},
                      systems={11: ["Retrieval & recall"]})
    line = next(ln for ln in out.splitlines() if "#11" in ln)
    assert '[snippet · Retrieval & recall] "embed_x"' in line
    assert TRIGGER[:40] not in line
    assert ">   ↳ the matched section" in out


@pytest.mark.asyncio
async def test_a_seen_record_is_a_pointer_not_a_copy():
    title = f"embed_x — {TRIGGER}"
    hits = [(0.8, fake_note(id=11, title=title, note_type="snippet",
                            data={"name": "embed_x"}, user_id=1))]
    out = await _menu(hits, [11], {11: {"index": 1, "text": f"{title}\nthe matched section"}})
    line = next(ln for ln in out.splitlines() if "#11" in ln)
    assert line == "> - #11 [snippet · seen] embed_x"
    assert "↳" not in out