FabledScribe/tests/test_autoinject_context.py

"""The auto-inject query carries the conversation beside a thin prompt (#4364).

The notes arm used to retrieve against the operator's typed words alone, and a
mid-session follow-up — "yes do that", "now fix the filter" — names nothing a
record can match. The hook now 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 scribe.services.plugin_context import (
    _AUTOINJECT_CONTEXT_MAX,
    _AUTOINJECT_CONTEXT_PROMPT_MAX,
    _autoinject_query,
    _menu_name,
    _menu_passage,
)
from tests.helpers import fake_note, need_tools

DEFS = Path(__file__).resolve().parents[1] / "plugin" / "hooks" / "scribe_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


# --- 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}"\nscribe_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 (#4364) ----------------------------------------
#
# 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 had gone 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 scribe.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