FabledScribe/tests/test_write_path_trigger.py

"""Tests for the write-path trigger (#2082) — prior art at the moment code is written.

Covers the service (`build_write_path_hint`): the two arms and their precedence,
the anti-bloat gates carried over from milestone 93, telemetry under its own
source, and the two ways this must stay silent. Plus the plugin hook contract —
a PreToolUse hook that returns a permission decision would be able to block the
operator's edit, which this feature must never do.
"""
import json
import re
import subprocess
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch

import pytest

PLUGIN = Path(__file__).resolve().parents[1] / "plugin"
HOOK = PLUGIN / "hooks" / "scribe_prior_art.sh"


def _snippet_item(nid, title, user_id=1):
    """A row as services.snippets.list_snippets returns it (a dict, not a Note)."""
    return {"id": nid, "title": title, "user_id": user_id, "note_type": "snippet"}


def _note(nid, title, user_id=1):
    n = MagicMock()
    n.id, n.title, n.user_id = nid, title, user_id
    return n


def _cfg(**over):
    base = {"enabled": True, "threshold": 0.55, "top_k": 3}
    base.update(over)
    return base


# --- silence: the common case ------------------------------------------------

@pytest.mark.asyncio
async def test_disabled_returns_empty_and_touches_nothing():
    from scribe.services import plugin_context as pc
    search, listing = AsyncMock(), AsyncMock()
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(enabled=False))), \
         patch.object(pc.snippets_svc, "list_snippets", listing), \
         patch.object(pc, "semantic_search_notes", search):
        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
    assert out["context"] == "" and out["note_ids"] == []
    listing.assert_not_called()
    search.assert_not_called()


@pytest.mark.asyncio
async def test_no_path_returns_empty():
    """Nothing to look up by place, and the semantic arm alone isn't this feature."""
    from scribe.services import plugin_context as pc
    search = AsyncMock()
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc, "semantic_search_notes", search):
        out = await pc.build_write_path_hint(1, "   ", code="def f(): ...")
    assert out["context"] == ""
    search.assert_not_called()


@pytest.mark.asyncio
async def test_nothing_recorded_is_silent():
    from scribe.services import plugin_context as pc
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
         patch.object(pc, "record_retrieval", MagicMock()):
        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
    assert out["context"] == "" and out["note_ids"] == []


# --- arm 1: by place ---------------------------------------------------------

@pytest.mark.asyncio
async def test_snippet_at_this_path_is_surfaced_without_a_score():
    """A snippet recorded HERE is prior art by definition, not by resemblance —
    it must not be subject to the similarity threshold."""
    from scribe.services import plugin_context as pc
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets",
                      AsyncMock(return_value=([_snippet_item(12, "debounce — rate-limit a callback")], 1))), \
         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
    assert out["note_ids"] == [12]
    assert '#12 [here] "debounce — rate-limit a callback"' in out["context"]
    assert "get_snippet(id)" in out["context"]
    assert "src/x.py" in out["context"]


@pytest.mark.asyncio
async def test_directory_is_only_consulted_when_the_file_leaves_room():
    from scribe.services import plugin_context as pc
    calls = []

    async def _listing(uid, **kw):
        calls.append(kw["path"])
        if kw["path"] == "src/lib/x.py":
            return ([_snippet_item(i, f"s{i}") for i in (1, 2, 3)], 3)
        return ([_snippet_item(9, "nearby")], 1)

    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
         patch.object(pc.snippets_svc, "list_snippets", _listing), \
         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/lib/x.py", code="x")
    # The file alone filled the budget, so the directory was never queried.
    assert calls == ["src/lib/x.py"]
    assert out["note_ids"] == [1, 2, 3]


@pytest.mark.asyncio
async def test_directory_hits_are_marked_nearby_not_here():
    from scribe.services import plugin_context as pc

    async def _listing(uid, **kw):
        if kw["path"] == "src/lib":
            return ([_snippet_item(9, "sibling helper")], 1)
        return ([], 0)

    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets", _listing), \
         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/lib/x.py", code="x")
    assert '#9 [nearby] "sibling helper"' in out["context"]


@pytest.mark.asyncio
async def test_a_failing_location_lookup_does_not_sink_the_hint():
    """The place arm is best-effort — a broken query must not cost the semantic one."""
    from scribe.services import plugin_context as pc
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(side_effect=RuntimeError("boom"))), \
         patch.object(pc, "semantic_search_notes",
                      AsyncMock(return_value=[(0.80, _note(5, "throttle — …"))])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
    assert out["note_ids"] == [5]


# --- arm 2: by meaning, under the milestone-93 gates -------------------------

@pytest.mark.asyncio
async def test_semantic_arm_is_snippet_only_and_browse_scoped():
    from scribe.services import plugin_context as pc
    search = AsyncMock(return_value=[])
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc, "semantic_search_notes", search), \
         patch.object(pc, "record_retrieval", MagicMock()):
        await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
    kwargs = search.await_args.kwargs
    # Prior art means snippets — not the dev-log that happens to mention one.
    assert kwargs["note_type"] == "snippet"
    # Nobody asked for this, so it must not reach a one-to-one direct share.
    assert kwargs["scope"] == "browse"


@pytest.mark.asyncio
async def test_margin_gate_applies_to_the_semantic_arm():
    from scribe.services import plugin_context as pc
    hits = [(0.80, _note(11, "near")), (0.74, _note(22, "alsoNear")), (0.61, _note(33, "far"))]
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=5))), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=hits)), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/x.py", code="def f(): ...")
    assert out["note_ids"] == [11, 22]
    assert "#33" not in out["context"]


@pytest.mark.asyncio
async def test_no_code_means_no_semantic_arm():
    """An Edit whose payload we couldn't read still gets the location answer."""
    from scribe.services import plugin_context as pc
    search = AsyncMock()
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets",
                      AsyncMock(return_value=([_snippet_item(12, "here helper")], 1))), \
         patch.object(pc, "semantic_search_notes", search), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/x.py", code="")
    assert out["note_ids"] == [12]
    search.assert_not_called()


@pytest.mark.asyncio
async def test_place_beats_meaning_and_the_cap_covers_both_arms():
    from scribe.services import plugin_context as pc

    async def _listing(uid, **kw):
        if kw["path"] == "src/x.py":
            return ([_snippet_item(1, "placed")], 1)
        return ([], 0)

    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=2))), \
         patch.object(pc.snippets_svc, "list_snippets", _listing), \
         patch.object(pc, "semantic_search_notes",
                      AsyncMock(return_value=[(0.9, _note(7, "scored"))])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/x.py", code="x")
    # Location hit first; the whole menu stays within top_k.
    assert out["note_ids"] == [1, 7]
    assert len(out["note_ids"]) <= 2
    assert out["context"].index("#1") < out["context"].index("#7")


@pytest.mark.asyncio
async def test_semantic_arm_only_asks_for_the_budget_the_place_arm_left():
    from scribe.services import plugin_context as pc
    search = AsyncMock(return_value=[])

    async def _listing(uid, **kw):
        if kw["path"] == "src/x.py":
            return ([_snippet_item(1, "placed")], 1)
        return ([], 0)

    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
         patch.object(pc.snippets_svc, "list_snippets", _listing), \
         patch.object(pc, "semantic_search_notes", search), \
         patch.object(pc, "record_retrieval", MagicMock()):
        await pc.build_write_path_hint(1, "src/x.py", code="x")
    assert search.await_args.kwargs["limit"] == 2


@pytest.mark.asyncio
async def test_session_dedup_excludes_ids_from_both_arms():
    from scribe.services import plugin_context as pc
    search = AsyncMock(return_value=[])
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets",
                      AsyncMock(return_value=([_snippet_item(12, "already shown")], 1))), \
         patch.object(pc, "semantic_search_notes", search), \
         patch.object(pc, "record_retrieval", MagicMock()):
        out = await pc.build_write_path_hint(1, "src/x.py", code="x", exclude_ids=[12])
    # The location hit was already surfaced this session → dropped, not repeated.
    assert out["note_ids"] == []
    assert 12 in search.await_args.kwargs["exclude_ids"]


# --- attribution + telemetry -------------------------------------------------

@pytest.mark.asyncio
async def test_another_users_snippet_is_attributed():
    """Rule #47 / the #2084 lesson: an unattributed line reads as your own record."""
    from scribe.services import plugin_context as pc
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets",
                      AsyncMock(return_value=([_snippet_item(12, "theirs", user_id=42)], 1))), \
         patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={42: "alex"})):
        out = await pc.build_write_path_hint(1, "src/x.py", code="x")
    assert "shared by alex, treat as a suggestion" in out["context"]


@pytest.mark.asyncio
async def test_telemetry_uses_its_own_source():
    """Separate source is what lets this surface be tuned apart from auto_inject."""
    from scribe.services import plugin_context as pc
    rec = MagicMock()
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg())), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc, "semantic_search_notes",
                      AsyncMock(return_value=[(0.9, _note(7, "scored"))])), \
         patch.object(pc, "record_retrieval", rec), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        await pc.build_write_path_hint(1, "src/x.py", code="x", project_id=4)
    rec.assert_called_once()
    assert rec.call_args.kwargs["source"] == "write_path"
    assert rec.call_args.kwargs["source"] != "auto_inject"
    assert rec.call_args.kwargs["project_id"] == 4


@pytest.mark.asyncio
async def test_config_has_its_own_switch_but_shares_the_gates():
    from scribe.services import plugin_context as pc
    stored = {pc.WRITEPATH_ENABLED_KEY: "false"}
    with patch.object(pc, "get_setting",
                      AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
        cfg = await pc.get_writepath_config(1)
    # Its own switch is off while auto-inject stays on...
    assert cfg["enabled"] is False
    # ...and the gates are auto-inject's, not a second set that could drift.
    assert cfg["threshold"] == pc.AUTOINJECT_DEFAULT_THRESHOLD
    assert cfg["top_k"] == pc.AUTOINJECT_DEFAULT_TOP_K


# --- the route between them --------------------------------------------------

def test_route_reads_every_arg_the_hook_sends():
    """Rule #33: the hook's query string and the route's reader are one contract,
    and the hook is the only caller — a rename on either side is silent."""
    import inspect

    from scribe.routes import plugin as routes
    src = inspect.getsource(routes.write_path_prior_art)
    for arg in ("path", "code", "repo", "project_id", "exclude_ids"):
        assert f'request.args.get("{arg}"' in src, f"route ignores {arg}"

    hook = HOOK.read_text()
    for arg in ("path=", "code=", "repo=", "exclude_ids="):
        assert arg in hook, f"hook never sends {arg}"


def test_route_resolves_repo_to_a_project_not_to_a_location_filter():
    """A snippet's `repo` is a label the operator typed; the hook sends a git
    remote. Matching one against the other would silently return nothing, so the
    remote may only reach resolve_project."""
    import inspect

    from scribe.routes import plugin as routes
    src = inspect.getsource(routes.write_path_prior_art)
    assert "resolve_project" in src
    assert "repo=repo" not in src


# --- the plugin surface ------------------------------------------------------

def test_hook_is_registered_for_write_and_edit():
    cfg = json.loads((PLUGIN / "hooks" / "hooks.json").read_text())
    entries = cfg["hooks"]["PreToolUse"]
    assert any(
        "Write" in e.get("matcher", "") and "Edit" in e.get("matcher", "")
        and any("scribe_prior_art.sh" in h["command"] for h in e["hooks"])
        for e in entries
    )


def test_hook_never_returns_a_permission_decision():
    """The load-bearing property: this hook informs a write, it cannot stop one.
    A permissionDecision of deny/ask would put a recall aid in the way of the
    operator's work."""
    src = HOOK.read_text()
    assert "additionalContext" in src
    code_lines = [ln for ln in src.splitlines() if not ln.lstrip().startswith("#")]
    assert not any("permissionDecision" in ln for ln in code_lines)


def test_hook_is_executable_and_shell_valid():
    assert HOOK.stat().st_mode & 0o111, "hook must be executable"
    subprocess.run(["bash", "-n", str(HOOK)], check=True)


def test_hook_reads_both_write_and_edit_payload_shapes():
    """Write and Edit name the payload differently, and the names have changed
    across Claude Code versions — read whichever is present."""
    src = HOOK.read_text()
    for field in ("content", "file_content", "new_string", "new_str"):
        assert f".tool_input.{field}" in src
    assert ".tool_input.file_path" in src


def test_hook_stays_a_get_so_a_read_scoped_key_works():
    """Every other plugin hook works with a read-scoped key; a POST would demand
    write scope (see auth.py) and silently break those installs."""
    src = HOOK.read_text()
    assert "-X POST" not in src and "--data" not in src
    assert "/api/plugin/prior-art?" in src


def test_hook_exits_silently_when_unconfigured():
    """An install with no Scribe URL/token must produce no output at all."""
    out = subprocess.run(
        ["bash", str(HOOK)],
        input=json.dumps({
            "session_id": "s1", "cwd": "/tmp", "tool_name": "Write",
            "tool_input": {"file_path": "/tmp/x.py", "content": "def f(): ..."},
        }),
        capture_output=True, text=True,
        env={"PATH": "/usr/bin:/bin", "SCRIBE_URL": "", "SCRIBE_TOKEN": ""},
    )
    assert out.returncode == 0
    assert out.stdout.strip() == ""


def test_hook_skips_prose_and_data_files():
    """No round-trip for a markdown edit — the server would return nothing anyway."""
    src = HOOK.read_text()
    skip = re.search(r"case \"\$file_path\" in\n(.*?)esac", src, re.S)
    assert skip, "expected an extension skip list"
    for ext in ("*.md", "*.json", "*.lock", "*.png"):
        assert ext in skip.group(1)
    # Config formats are deliberately NOT skipped — a workflow file is reusable.
    assert "*.yml" not in skip.group(1)


def test_plugin_version_bumped_with_the_hook():
    """The #1040 lesson: a plugin change clients can't see is a change that didn't
    ship."""
    manifest = json.loads((PLUGIN / ".claude-plugin" / "plugin.json").read_text())
    version = tuple(int(p) for p in manifest["version"].split("."))
    assert version >= (0, 1, 18)