"""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
from tests.helpers import fake_note, http_sink

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 _cfg(**over):
    # `rule_threshold` is the standing-rule arm's own bar (milestone 333 step
    # 4). It belongs in the stand-in even though most tests here never reach
    # that arm: the arm reads it while BUILDING its search arguments, so a
    # missing key raises inside its fail-open except and turns the arm into a
    # silent no-op — which is indistinguishable from it working and finding
    # nothing.
    base = {"enabled": True, "threshold": 0.68, "top_k": 3, "rule_threshold": 0.72}
    base.update(over)
    return base


# The semantic arm ignores payloads carrying less than WRITEPATH_MIN_CODE_CHARS
# of non-whitespace substance (#2223), so any test that expects it to RUN has to
# pass something a real Edit would plausibly contain. `code="x"` used to be the
# fixture here, which meant the gate tests were passing payloads that in
# production are exactly the noise the floor now drops.
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
'''


# --- 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=REAL_CODE)
    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=REAL_CODE)
    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=REAL_CODE)
    assert out["context"] == "" and out["note_ids"] == []


# --- the sync class, and arm 1 by place --------------------------------------

@pytest.mark.asyncio
async def test_snippet_at_this_path_becomes_the_sync_nudge():
    """A snippet recorded AT the exact file is not a reuse suggestion — it IS
    the record of the file being edited (#2708). It surfaces without a score,
    framed as "updating the record is part of this edit", and is reported in
    sync_note_ids so the hook feeds the sync dedup channel."""
    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=REAL_CODE)
    assert out["note_ids"] == [12]
    assert out["sync_note_ids"] == [12]
    assert '#12 [records this file] "debounce — rate-limit a callback"' in out["context"]
    # The load-bearing sentence: the record's freshness belongs to this edit.
    assert "update_snippet" in out["context"]
    assert "verify_snippet" in out["context"]
    assert "src/x.py" in out["context"]


@pytest.mark.asyncio
async def test_reuse_dedup_never_suppresses_the_sync_nudge():
    """THE bug #2708 names: a title surfaced as a reuse hint twenty turns ago
    landed in exclude_ids — and then silenced "you are editing the recorded
    file right now", the one claim that must fire at the moment of change. The
    sync class dedups only against its own channel."""
    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, "seen as reuse already")], 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=REAL_CODE, exclude_ids=[12],
        )
    assert out["sync_note_ids"] == [12]
    assert "[records this file]" in out["context"]


@pytest.mark.asyncio
async def test_sync_dedup_silences_a_sync_nudge_already_shown():
    """Once the session has been told "snippet #12 records this file", every
    further edit of the same file stays quiet about it — on every arm: the
    directory query and the semantic arm would both re-surface it otherwise."""
    from scribe.services import plugin_context as pc

    async def _listing(uid, **kw):
        # Recorded at the exact file, so it matches the file AND dir queries.
        return ([_snippet_item(12, "already nudged")], 1)

    search = AsyncMock(return_value=[])
    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", search), \
         patch.object(pc, "record_retrieval", MagicMock()):
        out = await pc.build_write_path_hint(
            1, "src/x.py", code=REAL_CODE, exclude_sync_ids=[12],
        )
    assert out["note_ids"] == [] and out["sync_note_ids"] == []
    assert "#12" not in out["context"]
    assert 12 in search.await_args.kwargs["exclude_ids"]


@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=REAL_CODE)
    # 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=REAL_CODE)
    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, fake_note(id=5, title="throttle — …", user_id=1, note_type="snippet"))])), \
         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=REAL_CODE)
    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=REAL_CODE)
    kwargs = search.await_args.kwargs
    # Prior art is snippets AND recorded experience (#2246) — an issue saying
    # "we tried this and it broke" belongs here. What stays out is the open
    # to-do list, which resembles the code and answers nothing.
    assert kwargs["note_type"] == ("snippet", "note")
    assert kwargs["task_kind"] == "issue"
    # 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, fake_note(id=11, title="near", user_id=1, note_type="snippet")), (0.74, fake_note(id=22, title="alsoNear", user_id=1, note_type="snippet")), (0.61, fake_note(id=33, title="far", user_id=1, note_type="snippet"))]
    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=REAL_CODE)
    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, fake_note(id=7, title="scored", user_id=1, note_type="snippet"))])), \
         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=REAL_CODE)
    # 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=REAL_CODE)
    assert search.await_args.kwargs["limit"] == 2


@pytest.mark.asyncio
async def test_session_dedup_excludes_ids_from_the_reuse_arms():
    """exclude_ids governs the REUSE classes — nearby and semantic. (The sync
    class has its own channel; see the tests above.)"""
    from scribe.services import plugin_context as pc

    async def _listing(uid, **kw):
        # Recorded at a SIBLING file, so it reaches only the directory query.
        if kw["path"] == "src":
            return ([_snippet_item(12, "already shown")], 1)
        return ([], 0)

    search = AsyncMock(return_value=[])
    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", search), \
         patch.object(pc, "record_retrieval", MagicMock()):
        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, exclude_ids=[12])
    # The nearby 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=REAL_CODE)
    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, fake_note(id=7, title="scored", user_id=1, note_type="snippet"))])), \
         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=REAL_CODE, project_id=4)
    sources = [c.kwargs["source"] for c in rec.call_args_list]
    assert sources.count("write_path") == 1, sources
    assert "auto_inject" not in sources

    note_arm = next(c for c in rec.call_args_list if c.kwargs["source"] == "write_path")
    assert note_arm.kwargs["project_id"] == 4

    # The rule arm rides along on the same hint and logs its own call even when
    # it finds nothing (#3497). This assertion used to be `assert_called_once`,
    # which passed only because that row was never written — the test encoded
    # the defect. The second row is the point of having two sources.
    assert "write_path_rule" in sources, sources


@pytest.mark.asyncio
async def test_sync_surfacing_is_measured_under_its_own_usage_source():
    """The sync class's pull-through rate is #2708's scoreboard — whether
    sessions actually update the record when told they're editing it. Folding
    it into write_path_place would make that unmeasurable."""
    from scribe.services import plugin_context as pc

    async def _listing(uid, **kw):
        if kw["path"] == "src/x.py":
            return ([_snippet_item(12, "records me")], 1)
        return ([_snippet_item(9, "sibling")], 1)

    surfaced = MagicMock()
    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, "record_surfaced", surfaced), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
    by_source = {
        c.kwargs["source"]: c.kwargs["note_ids"] for c in surfaced.call_args_list
    }
    assert by_source["write_path_sync"] == [12]
    assert by_source["write_path_place"] == [9]


@pytest.mark.asyncio
async def test_config_has_its_own_switch_and_threshold_but_shares_top_k():
    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
    # ...its threshold is its OWN and stricter than auto-inject's, because code
    # embeddings have a higher similarity floor than prose (#2223). Sharing 0.55
    # made unrelated code — including `x = 1` at 0.58 — clear the bar.
    assert cfg["threshold"] == pc.WRITEPATH_DEFAULT_THRESHOLD
    assert cfg["threshold"] > pc.AUTOINJECT_DEFAULT_THRESHOLD
    # ...and top_k is still shared: "how many titles at once" means the same
    # thing on both surfaces.
    assert cfg["top_k"] == pc.AUTOINJECT_DEFAULT_TOP_K


@pytest.mark.asyncio
async def test_writepath_threshold_is_operator_tunable_and_clamped():
    """Rule #25 — it's a product knob, so it has to be settable, and a typo in
    the settings box must not be able to open the gate to everything."""
    from scribe.services import plugin_context as pc

    async def _cfg_with(raw):
        stored = {pc.WRITEPATH_THRESHOLD_KEY: raw}
        with patch.object(pc, "get_setting",
                          AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
            return await pc.get_writepath_config(1)

    assert (await _cfg_with("0.9"))["threshold"] == 0.9
    assert (await _cfg_with("5"))["threshold"] == 1.0        # clamped up
    assert (await _cfg_with("-3"))["threshold"] == 0.0       # clamped down
    # Garbage falls back to the default rather than to 0.0, which would surface
    # every snippet in the corpus on every edit.
    assert (await _cfg_with("banana"))["threshold"] == pc.WRITEPATH_DEFAULT_THRESHOLD


@pytest.mark.asyncio
async def test_the_rule_arm_has_its_own_tunable_bar():
    """Rule #25 again, for the THIRD corpus (milestone 333 step 4).

    Separate from the code threshold above and separately settable, because the
    two are measured against different things: 0.68 was derived from code
    against note PROSE (#2223), and rules are short imperative technical
    English — a more homogeneous corpus whose noise floor sits higher.
    """
    from scribe.services import plugin_context as pc

    async def _cfg_with(raw):
        stored = {pc.RULEHINT_THRESHOLD_KEY: raw}
        with patch.object(pc, "get_setting",
                          AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
            return await pc.get_writepath_config(1)

    assert (await _cfg_with("0.8"))["rule_threshold"] == 0.8
    assert (await _cfg_with("5"))["rule_threshold"] == 1.0
    assert (await _cfg_with("-3"))["rule_threshold"] == 0.0
    # Garbage falls back to the default, not to 0.0 — which on THIS arm would
    # attach a standing rule to every write in the session.
    assert (await _cfg_with("banana"))["rule_threshold"] == pc.RULEHINT_DEFAULT_THRESHOLD


@pytest.mark.asyncio
async def test_the_two_write_path_bars_are_independent():
    """The split, asserted. Setting one must not move the other — the failure
    that would silently undo this step is a config assembler that reads one key
    into both fields."""
    from scribe.services import plugin_context as pc

    stored = {pc.WRITEPATH_THRESHOLD_KEY: "0.90", pc.RULEHINT_THRESHOLD_KEY: "0.61"}
    with patch.object(pc, "get_setting",
                      AsyncMock(side_effect=lambda uid, k, d: stored.get(k, d))):
        cfg = await pc.get_writepath_config(1)

    assert cfg["threshold"] == 0.90
    assert cfg["rule_threshold"] == 0.61


def test_the_rule_bar_defaults_above_the_code_bar():
    """Not a number check — a DIRECTION check, and the only part of the default
    that is defensible without one instance's histogram (rule 115).

    The eligible rule corpus is orders of magnitude smaller than the note
    corpus, so a top-k over it always returns something and a bar calibrated
    for best-of-thousands is cleared by best-of-forty as arithmetic. Rules are
    also more homogeneous than note prose, so their noise floor is higher. Both
    facts point the same way: this bar must sit ABOVE the one it inherited.

    Pinned as an inequality so tuning the value stays free while inverting the
    relationship — which would silently reinstate #3311 — does not.
    """
    from scribe.services import plugin_context as pc

    assert pc.RULEHINT_DEFAULT_THRESHOLD > pc.WRITEPATH_DEFAULT_THRESHOLD


def test_the_rule_arm_asks_for_one_rule_not_two():
    """With a corpus this small, top-k does as much damage as the threshold:
    k=2 over a few dozen candidates means the second line is almost always the
    second-best noise, carrying the same confident framing as the first."""
    from scribe.services import plugin_context as pc

    assert pc.RULEHINT_LIMIT == 1


# --- the minimum-substance floor on the semantic arm (#2223) ------------------

@pytest.mark.asyncio
async def test_trivial_payload_skips_the_semantic_arm_entirely():
    """`x = 1` scored 0.58 against an unrelated helper on the live instance — the
    embedder's floor for "some code", not a resemblance. A payload too small to
    carry meaning must not reach the embedder at all, and must not cost a
    retrieval_logs row either."""
    from scribe.services import plugin_context as pc
    search, rec = AsyncMock(return_value=[]), 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", search), \
         patch.object(pc, "record_retrieval", rec):
        out = await pc.build_write_path_hint(1, "src/x.py", code="x = 1")
    assert out["context"] == "" and out["note_ids"] == []
    search.assert_not_called()
    rec.assert_not_called()


@pytest.mark.asyncio
async def test_the_floor_counts_substance_not_indentation():
    """Code is indentation-heavy, so raw length would let a deeply-nested
    one-liner through on padding alone."""
    from scribe.services import plugin_context as pc
    padded = " " * 400 + "return None"   # long, but almost no substance
    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=padded)
    assert len(padded) > pc.WRITEPATH_MIN_CODE_CHARS   # would pass a raw-length check
    search.assert_not_called()


@pytest.mark.asyncio
async def test_the_floor_does_not_block_the_place_arm():
    """The floor is a statement about the semantic query, not about the edit. A
    snippet recorded at this exact path is prior art however small the change."""
    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, "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="x = 1")
    assert out["note_ids"] == [12]
    search.assert_not_called()


@pytest.mark.asyncio
async def test_a_real_helper_clears_the_floor():
    """The floor errs toward keeping recall — anything that plausibly IS a
    reusable helper has to get through, or the feature stops working."""
    from scribe.services import plugin_context as pc
    search = AsyncMock(return_value=[(0.80, fake_note(id=5, title="debounce — …", user_id=1, note_type="snippet"))])
    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()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
    assert out["note_ids"] == [5]
    search.assert_called_once()


def test_the_floor_stays_below_the_smallest_plausible_helper():
    """A guard on the constant itself: raising it far enough to reject a real
    one-line helper would silently gut the arm rather than sharpen it."""
    from scribe.services import plugin_context as pc
    smallest_real_helper = 'def slug(t): return re.sub(r"[^a-z0-9]+", "-", t.lower()).strip("-")'
    substance = len("".join(smallest_real_helper.split()))
    assert pc.WRITEPATH_MIN_CODE_CHARS < substance
    # And still strictly above a degenerate one-liner.
    assert pc.WRITEPATH_MIN_CODE_CHARS > len("".join("x = 1".split()))


# --- concept extraction: query by what the code IS FOR, not by the code (#2242) ---

def test_concept_query_prefers_the_name_and_docstring():
    """Snippet documents are prose-forward — `when_to_use` appears in both the
    title and the body — so a concept query out-scores the code itself (0.823 vs
    0.743 measured against #2222). The shape mirrors a snippet's own title."""
    from scribe.services.plugin_context import concept_query
    q = concept_query(
        'def collapse_into_clusters(edges):\n'
        '    """Collapse similar-pairs into connected components."""\n'
        '    parent = {}\n'
        '    return parent\n'
    )
    assert q == "collapse_into_clusters(edges) — Collapse similar-pairs into connected components."


def test_concept_query_reads_jsdoc_and_the_const_arrow_form():
    """The dominant shape in this repo's frontend isn't `function foo()`."""
    from scribe.services.plugin_context import concept_query
    q = concept_query(
        "/**\n"
        " * Rate-limit a callback so it fires once after the last call.\n"
        " */\n"
        "export const debounce = (fn: Fn, wait = 250) => {\n"
        "  let t: number | undefined;\n"
        "};\n"
    )
    assert q == "debounce(fn: Fn, wait = 250) — Rate-limit a callback so it fires once after the last call."


def test_concept_query_does_not_let_a_shebang_become_the_description():
    """`#!` matches the leading-comment pattern but says nothing about the code."""
    from scribe.services.plugin_context import concept_query
    q = concept_query(
        "#!/usr/bin/env bash\n"
        "# Resolve the repo remote and url-encode it for the plugin API.\n"
        "encode_remote() {\n"
        "  git remote get-url origin\n"
        "}\n"
    )
    assert q == "encode_remote() — Resolve the repo remote and url-encode it for the plugin API."
    assert "usr/bin/env" not in q


def test_concept_query_falls_back_when_there_is_no_doc():
    """MEASURED RULE, not a taste call. An identifier alone scored 0.671 against
    #2222 where the full code body scored 0.743 — same separation from the noise
    floor (0.113), but below the 0.68 bar, so preferring the bare name would turn
    a comfortable hit into a miss. Undocumented code keeps the raw payload."""
    from scribe.services.plugin_context import concept_query
    assert concept_query(
        "def collapse_into_clusters(edges):\n"
        "    parent = {}\n"
        "    return parent\n"
    ) == ""


def test_concept_query_falls_back_on_a_doc_that_says_nothing():
    from scribe.services.plugin_context import concept_query
    assert concept_query('def f(x):\n    """TODO"""\n    return x\n') == ""


def test_concept_query_falls_back_when_nothing_parses():
    """A Vue SFC has neither a recognised declaration nor a doc block. Every
    unhandled language must degrade to exactly the pre-#2242 behaviour."""
    from scribe.services.plugin_context import concept_query
    assert concept_query(
        '<script setup lang="ts">\n'
        "const props = defineProps<{ modelValue: string }>()\n"
        "</script>\n"
    ) == ""


def test_concept_query_accepts_a_doc_without_a_declaration():
    """A module docstring is already the concept; it needs no signature."""
    from scribe.services.plugin_context import concept_query
    q = concept_query('"""Turn a raw git remote into the project it belongs to."""\nMAPPING = {}\n')
    assert q == "Turn a raw git remote into the project it belongs to."


def test_concept_query_caps_what_it_collects():
    """A 40-function Write must not become a wall of signatures, and a long module
    docstring must not drown out the declaration."""
    from scribe.services import plugin_context as pc
    many = '"""Utilities."""\n' + "".join(
        f"def helper_{i}(a, b):\n    return a\n" for i in range(12)
    )
    q = pc.concept_query(many)
    assert q.count("helper_") == pc._CONCEPT_MAX_DECLS

    long_doc = '"""' + ("word " * 400) + '"""\ndef f(a):\n    return a\n'
    assert len(pc.concept_query(long_doc)) <= pc._CONCEPT_MAX_DOC_CHARS + 64


@pytest.mark.asyncio
async def test_semantic_arm_queries_the_concept_not_the_raw_payload():
    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=REAL_CODE)
    sent = search.await_args.args[1] if len(search.await_args.args) > 1 else search.await_args.kwargs["query"]
    assert sent.startswith("debounce(fn, wait=0.25)")
    assert "Rate-limit a callback so it fires once after the last call." in sent
    assert "nonlocal timer" not in sent   # the implementation body is gone


@pytest.mark.asyncio
async def test_telemetry_records_the_query_actually_sent():
    """retrieval_logs is what the threshold gets tuned from, so it has to hold the
    concept query — logging the raw code would make the scores uninterpretable."""
    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=[])), \
         patch.object(pc, "record_retrieval", rec):
        await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
    logged = rec.call_args.kwargs["query"]
    assert logged.startswith("debounce(fn, wait=0.25)")
    assert "nonlocal timer" not in logged


@pytest.mark.asyncio
async def test_the_floor_judges_the_raw_payload_not_the_concept():
    """ORDER OF OPERATIONS. The floor asks "is this a helper being written?" of the
    RAW payload; the rewrite happens after. A concept query is allowed to be
    shorter than the floor — that's the whole point, since the best queries are
    short — but a sub-floor payload must stay silent even when it has a docstring."""
    from scribe.services import plugin_context as pc

    # Clears the floor, and its concept is deliberately SHORTER than the floor.
    verbose = (
        'def slugify(text):\n'
        '    """Turn text into a url slug."""\n'
        '    out = re.sub(r"[^a-z0-9]+", "-", text.lower())\n'
        '    return out.strip("-")\n'
    )
    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=verbose)
    concept = pc.concept_query(verbose)
    assert len("".join(concept.split())) < pc.WRITEPATH_MIN_CODE_CHARS
    search.assert_called_once()   # ...and it still searched

    # Under the floor, docstring notwithstanding.
    tiny = 'def f(x):\n    """Slug it."""\n    return x\n'
    search2 = 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", search2), \
         patch.object(pc, "record_retrieval", MagicMock()):
        await pc.build_write_path_hint(1, "src/x.py", code=tiny)
    assert len("".join(tiny.split())) < pc.WRITEPATH_MIN_CODE_CHARS
    search2.assert_not_called()


# --- cross-language prior art is disclosed, not filtered (#2244) --------------

def test_language_for_path_and_canonicalisation():
    from scribe.services import plugin_context as pc
    assert pc._language_for_path("src/a/b.ts") == "typescript"
    assert pc._language_for_path("src/a/b.py") == "python"
    assert pc._language_for_path("ops/Dockerfile") == "dockerfile"
    assert pc._language_for_path("README") == ""
    assert pc._language_for_path("weird/thing.qqq") == ""
    # Operator-typed spellings that mean the same thing must compare equal.
    assert pc._canonical_language("Python3") == pc._canonical_language("py")
    assert pc._canonical_language("TSX") == pc._canonical_language("typescript")
    assert pc._canonical_language("C++") == "cpp"
    # Unknown-but-equal still compares equal, which is all this must get right.
    assert pc._canonical_language("Brainfuck") == pc._canonical_language("brainfuck")


def test_foreign_language_only_claims_a_mismatch_it_can_establish():
    from scribe.services import plugin_context as pc
    assert pc._foreign_language({"language": "python"}, "typescript") == "python"
    assert pc._foreign_language({"language": "py"}, "python") == ""       # same, aliased
    assert pc._foreign_language({"language": ""}, "typescript") == ""     # theirs unknown
    assert pc._foreign_language({"language": "python"}, "") == ""         # target unknown
    assert pc._foreign_language({}, "typescript") == ""                   # no field at all


@pytest.mark.asyncio
async def test_a_cross_language_hit_is_labelled_and_explained():
    """The fail state this closes: an unlabelled Python hit offered while writing
    TypeScript is either dismissed as irrelevant or pasted into the .ts file."""
    from scribe.services import plugin_context as pc
    note = fake_note(id=7, title="group_pairs — collapse related pairs into groups", user_id=1, note_type="snippet")
    note.data = {"language": "python"}
    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.72, note)])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "frontend/src/lib/cluster.ts", code=REAL_CODE)
    assert "[similar 0.72 · python]" in out["context"]
    assert "shape of a solution to adapt, not code to copy" in out["context"]
    # Disclosed, NOT filtered — a stricter bar for foreign hits would suppress
    # exactly the shape-borrowing this exists for.
    assert out["note_ids"] == [7]


@pytest.mark.asyncio
async def test_a_same_language_hit_is_not_labelled_and_gets_no_preamble():
    """The common case keeps a clean line; the explanation only appears when
    there is something on the menu it explains."""
    from scribe.services import plugin_context as pc
    note = fake_note(id=7, title="debounce — rate-limit a callback", user_id=1, note_type="snippet")
    note.data = {"language": "python"}
    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.72, note)])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "src/scribe/services/x.py", code=REAL_CODE)
    assert "[similar 0.72]" in out["context"]
    assert "·" not in out["context"]
    assert "shape of a solution" not in out["context"]


@pytest.mark.asyncio
async def test_an_unknown_target_extension_never_invents_a_mismatch():
    """A wrong "· python" tag is worse than no tag at all."""
    from scribe.services import plugin_context as pc
    note = fake_note(id=7, title="helper — does a thing", user_id=1, note_type="snippet")
    note.data = {"language": "python"}
    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.72, note)])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
        out = await pc.build_write_path_hint(1, "scripts/thing.qqq", code=REAL_CODE)
    assert "· python" not in out["context"]


def test_prior_art_line_keeps_language_and_attribution_together():
    from scribe.services.plugin_context import _prior_art_line
    item = {"id": 12, "title": "group_pairs — collapse pairs"}
    assert _prior_art_line(item, "similar 0.72", None, "python") == \
        '> - #12 [similar 0.72 · python] "group_pairs — collapse pairs"'
    both = _prior_art_line(item, "similar 0.72", "alex", "python")
    assert "· python" in both and "shared by alex" in both


# --- 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
    # The handler plus the query-scope helper it delegates repo/project_id
    # to — the contract is what the ROUTE MODULE reads, wherever it reads it.
    src = inspect.getsource(routes.write_path_prior_art) + inspect.getsource(
        routes._project_scope
    )
    for arg in ("path", "code", "repo", "project_id", "exclude_ids",
                "exclude_sync_ids", "shapes"):
        assert f'request.args.get("{arg}"' in src, f"route ignores {arg}"

    hook = HOOK.read_text()
    for arg in ("path=", "code=", "repo=", "exclude_ids=", "exclude_sync_ids=",
                "shapes="):
        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) + inspect.getsource(
        routes._project_scope
    )
    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. The list lives in the shared library (#2901) and the hook asks it."""
    lib = (PLUGIN / "hooks" / "scribe_defs.sh").read_text()
    skip = re.search(r"scribe_skip_path\(\) \{\n  case \"\$1\" in\n(.*?)esac", lib, re.S)
    assert skip, "expected an extension skip list in scribe_defs.sh"
    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)
    src = HOOK.read_text()
    assert 'scribe_skip_path "$file_path" && exit 0' in src
    assert '/scribe_defs.sh"' in src                   # sourced, not copied


def test_hook_keeps_sync_and_reuse_dedup_apart():
    """#2708's dedup audit, pinned: the hook holds TWO per-session id files and
    feeds each its own class — sync ids (snippets recording the edited file) to
    the sync file, the rest to the reuse file. One shared file is exactly the
    bug this replaced: a reuse hint early in the session silencing the record-
    sync nudge when the recorded file is edited later."""
    src = HOOK.read_text()
    assert ".sync.ids" in src                       # its own state file
    assert "exclude_sync_ids=" in src               # its own query channel
    # The reuse file must NOT swallow sync ids — the write-back subtracts them.
    assert "(.note_ids // []) - (.sync_note_ids // [])" in src
    assert "(.sync_note_ids // [])[]?" in src


def _hook_runtime_env():
    """Env for tests that need the hook's tools to actually RUN.

    The silence-contract tests above deliberately restrict PATH — the hook must
    exit quietly when its tools are missing, and asserting on empty output
    doesn't care why it was empty. These tests assert on CONTENT, so the tools
    must resolve wherever the image installed them; skip (don't fail) on an
    image that lacks them, because that image cannot exercise this behaviour
    at all.
    """
    import os
    import shutil

    for tool in ("git", "jq", "curl", "bash"):
        if shutil.which(tool) is None:
            pytest.skip(f"hook runtime tool {tool!r} not installed")
    return {"PATH": os.environ["PATH"],
            "SCRIBE_URL": "http://127.0.0.1:9", "SCRIBE_TOKEN": "t"}


def _dup_repo(tmp_path, env):
    repo = tmp_path / "repo"
    repo.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
    (repo / "a.py").write_text("def debounce(fn):\n    return fn\n")
    # git grep searches the index, so the existing copy must be staged.
    subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
    return repo


def _write_event(repo, session="s-nudge"):
    return json.dumps({
        "session_id": session, "cwd": str(repo), "tool_name": "Write",
        "tool_input": {"file_path": str(repo / "b.py"),
                       "content": "def debounce(fn):\n    return fn\n"},
    })


def test_hook_nudges_recording_when_copies_exist_but_nothing_is_recorded(tmp_path):
    """#2664: the local arm proves duplication; when Scribe ANSWERS that it has
    no record of it, the same context block must ask for create_snippet — the
    one moment the recording nudge is earned rather than noise."""
    with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, seen):
        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
        repo = _dup_repo(tmp_path, env)
        out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo),
                             capture_output=True, text=True, env=env)
    assert out.returncode == 0
    assert seen and seen[0]["path"] == ["b.py"]
    assert out.stdout.strip(), (
        "hook produced no output — the local arm should have found the "
        "staged duplicate and nudged"
    )
    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
    assert "already defined" in ctx          # the duplication finding
    assert "create_snippet" in ctx           # the recording ask riding it
    assert "did not answer" not in ctx


def test_hook_says_when_scribe_did_not_answer_once_per_outage(tmp_path):
    """#2932: a configured instance that does not answer (refused connection)
    is SAID — the write went unchecked — instead of the hook failing open in
    silence; the record nudge's "nothing recorded" claim is withheld. Once per
    outage: a second miss is quiet, an answer clears the marker, and the next
    miss speaks again. The marker is shared with the after-write hook."""
    env = _hook_runtime_env()                 # SCRIBE_URL → a refused port
    repo = _dup_repo(tmp_path, env)
    marker = tmp_path / "scribe-priorart" / "s-out.unreached"
    env["TMPDIR"] = str(tmp_path)
    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
                         capture_output=True, text=True, env=env)
    assert out.returncode == 0
    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
    assert "already defined" in ctx
    assert "Scribe did not answer the prior-art check for `b.py` within 5s" in ctx
    assert "UNCHECKED" in ctx and "list_shapes" in ctx
    assert "None of those existing copies is recorded" not in ctx
    assert marker.is_file()
    # Second miss inside the quiet window: local arm only.
    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
                         capture_output=True, text=True, env=env)
    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
    assert "already defined" in ctx and "did not answer" not in ctx
    # An answer clears the marker …
    with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, _seen):
        up = dict(env, SCRIBE_URL=f"http://127.0.0.1:{port}")
        subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
                       capture_output=True, text=True, env=up)
    assert not marker.exists()
    # … so the next outage is announced afresh.
    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out"),
                         capture_output=True, text=True, env=env)
    assert "did not answer" in json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
    # A write the hook had nothing local to say about still carries the line
    # (the line is the whole message then): a fresh session, no duplicate.
    (repo / "a.py").unlink()
    subprocess.run(["git", "add", "-A"], cwd=repo, check=True, env=env)
    out = subprocess.run(["bash", str(HOOK)], input=_write_event(repo, "s-out-2"),
                         capture_output=True, text=True, env=env)
    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
    assert ctx.startswith("> Scribe did not answer")


def test_hook_stays_quiet_about_recording_when_nothing_is_duplicated(tmp_path):
    """A brand-new helper with no other copies earns no nudge — a reflex that
    fires on every Write is one sessions learn to skip."""
    env = _hook_runtime_env()
    repo = tmp_path / "repo"
    repo.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
    out = subprocess.run(
        ["bash", str(HOOK)],
        input=json.dumps({
            "session_id": "s-quiet", "cwd": str(repo), "tool_name": "Write",
            "tool_input": {"file_path": str(repo / "b.py"),
                           "content": "def debounce(fn):\n    return fn\n"},
        }),
        capture_output=True, text=True, env=env,
    )
    assert out.returncode == 0
    assert "create_snippet" not in out.stdout


@pytest.mark.parametrize(
    ("fname", "definition"),
    [
        ("scanner.go", "func Resolve(x int) error {\n\treturn nil\n}\n"),
        ("scanner_m.go",
         "func (s *Scanner) Resolve(x int) error {\n\treturn nil\n}\n"),
        ("queue.kt", "suspend fun refreshQueue(id: Long) {\n}\n"),
        ("fetch.rs", "pub async fn fetch_all() -> u32 {\n    0\n}\n"),
        ("adapter.go", "type ForgeAdapter struct {\n\tname string\n}\n"),
    ],
    ids=["go-func", "go-method", "kotlin-fun", "rust-fn", "go-type"],
)
def test_local_arm_finds_duplicates_in_every_language_family(
    tmp_path, fname, definition
):
    """#2682: the definition detector must cover ALL code, not the languages of
    the repo it was born in. Its original CSS/JS/Python-only patterns silently
    amputated the local arm — and the #2664 recording nudge gated on it — for
    every Go/Kotlin/Rust project, which is exactly where the operator observed
    recording never happening. Each case stages an existing copy and writes the
    same definition to a second file; the hook must prove the duplication and
    ask for the record (the instance ANSWERS "nothing recorded" — since #2932
    an unanswered call withholds the nudge, so a sink stands in for it)."""
    with http_sink(b'{"context":"","note_ids":[],"sync_note_ids":[]}') as (port, _seen):
        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
        repo = tmp_path / "repo"
        repo.mkdir()
        subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
        (repo / fname).write_text(definition)
        subprocess.run(["git", "add", "."], cwd=repo, check=True, env=env)
        ext = fname.rsplit(".", 1)[1]
        out = subprocess.run(
            ["bash", str(HOOK)],
            input=json.dumps({
                "session_id": f"s-lang-{ext}", "cwd": str(repo),
                "tool_name": "Write",
                "tool_input": {"file_path": str(repo / f"copy.{ext}"),
                               "content": definition},
            }),
            capture_output=True, text=True, env=env,
        )
    assert out.returncode == 0
    assert out.stdout.strip(), (
        f"hook produced no output for {fname} — the local arm should have "
        f"found the staged duplicate definition"
    )
    ctx = json.loads(out.stdout)["hookSpecificOutput"]["additionalContext"]
    assert "already defined" in ctx
    assert "create_snippet" in ctx


# --- #2791: the write-path feed — hook evidence lands as ledger rows ----------


def _ts():
    from datetime import datetime, timezone
    return datetime.now(timezone.utc)


@pytest.mark.asyncio
async def test_stamping_needs_named_shapes_and_a_recent_pull():
    """Offered-but-ignored stamps nothing: without a PULLED event there is no
    evidence, and without the hook naming shapes there is nothing to stamp.
    Neither case may even read the pull stream."""
    from scribe.services import plugin_context as pc
    pulls = AsyncMock(return_value={})
    stamp = 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", AsyncMock(return_value=[])), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", pulls), \
         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp):
        out = await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, project_id=4)
        assert out["stamped"] == []
        pulls.assert_not_awaited()                       # no shapes → no read
        out = await pc.build_write_path_hint(
            1, "src/x.py", code=REAL_CODE, project_id=4,
            stamp_shapes=[("sym", "debounce")],
        )
    pulls.assert_awaited_once()
    stamp.assert_not_awaited()                           # shapes, but no pull
    assert out["stamped"] == []


@pytest.mark.asyncio
async def test_a_pulled_snippet_already_seen_is_evidence_not_menu():
    """The pulled-then-written flow IS the dedup-excluded flow: the hint offered
    #7 earlier (so it sits in exclude_ids), the session pulled it, and now
    writes code resembling it. #7 must be scored for this payload — and handed
    to the stamp as resemblance — without being re-listed in the menu."""
    from scribe.services import plugin_context as pc
    search = AsyncMock(return_value=[(0.91, fake_note(id=7, title="pulled", user_id=1, note_type="snippet")), (0.80, fake_note(id=8, title="fresh", user_id=1, note_type="snippet"))])
    stamp = AsyncMock(return_value=[{
        "path": "src/x.py", "symbol": "debounce", "kind": "sym",
        "snippet_id": 7, "reason": "hook: pulled #7; payload resembles it (0.91)",
    }])
    with patch.object(pc, "get_writepath_config", AsyncMock(return_value=_cfg(top_k=3))), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc, "semantic_search_notes", search), \
         patch.object(pc, "record_retrieval", MagicMock()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={7: _ts()})), \
         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances", stamp):
        out = await pc.build_write_path_hint(
            1, "src/x.py", code=REAL_CODE, project_id=4, exclude_ids=[7],
            stamp_shapes=[("sym", "debounce")], repo_key="git.example.com/a/b",
        )
    # The query kept #7 eligible (and widened the budget by one for it)...
    kw = search.call_args.kwargs
    assert 7 not in kw["exclude_ids"]
    assert kw["limit"] == 4
    # ...but the menu still honours the session dedup.
    assert out["note_ids"] == [8]
    assert "#7" not in "\n".join(
        line for line in out["context"].splitlines() if "[similar" in line
    )
    # The stamp saw the pull and the resemblance score for this payload.
    skw = stamp.call_args.kwargs
    assert skw["pulled"] == {7: skw["pulled"][7]}
    assert skw["resembles"] == {7: 0.91}
    assert skw["shapes"] == [("sym", "debounce")]
    assert skw["repo_key"] == "git.example.com/a/b"
    assert out["stamped"][0]["snippet_id"] == 7
    # And the session is told what landed, with the way to correct it.
    assert "Shape accounting" in out["context"]
    assert "`debounce` → instance of #7" in out["context"]
    assert "classify_shapes" in out["context"]


@pytest.mark.asyncio
async def test_a_stamp_renders_even_when_the_hint_is_otherwise_silent():
    """After dedup the common case is an empty hint; the stamp must still run
    and still be reported — silence about accounting is how hook rows would
    become invisible."""
    from scribe.services import plugin_context as pc
    stamped = [{"path": "web/b.css", "symbol": "btn-primary", "kind": "css",
                "snippet_id": 5, "reason": "hook: pulled #5; payload references `btn-primary`"}]
    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()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={5: _ts()})), \
         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances",
                      AsyncMock(return_value=stamped)):
        out = await pc.build_write_path_hint(
            1, "web/b.css", code=".btn-primary { color: red; }" * 4, project_id=4,
            stamp_shapes=[("css", "btn-primary")],
        )
    assert out["note_ids"] == []
    assert out["stamped"] == stamped
    assert "`.btn-primary` → instance of #5" in out["context"]


@pytest.mark.asyncio
async def test_a_failing_stamp_does_not_sink_the_hint():
    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, "records me")], 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={})), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={12: _ts()})), \
         patch.object(pc.shape_ledger_svc, "stamp_write_path_instances",
                      AsyncMock(side_effect=RuntimeError("ledger down"))):
        out = await pc.build_write_path_hint(
            1, "src/x.py", code=REAL_CODE, project_id=4, stamp_shapes=[("sym", "f")],
        )
    assert out["sync_note_ids"] == [12]
    assert out["stamped"] == []


def test_route_stamps_only_for_a_caller_allowed_to_write():
    """A read-scoped key gets the hint — every plugin hook works on a read key
    — but a GET must never change accounting for it. The route passes the
    shapes through only when the key is write-scoped (or it's a session)."""
    import inspect

    from scribe.routes import plugin as routes
    src = inspect.getsource(routes.write_path_prior_art)
    assert 'request.args.get("shapes"' in src
    assert '== "write"' in src
    assert "stamp_shapes=shapes if may_stamp else None" in src
    assert "normalize_repo_key(repo)" in src
    hook = HOOK.read_text()
    assert "&shapes=" in hook


def test_hook_does_not_name_a_type_import_specifier_as_a_shape(tmp_path):
    """#2904, the awk mirror of the server rule: `type Foo,` inside an import
    list is not a definition; `type Baz = …` on its own line is."""
    env = _hook_runtime_env()
    repo = tmp_path / "repo"
    repo.mkdir()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
    seen = _run_hook_against_sink(tmp_path, {
        "session_id": "s-type", "cwd": str(repo), "tool_name": "Write",
        "tool_input": {"file_path": str(repo / "x.ts"),
                       "content": 'import { type Foo, bar } from "./y";\n'
                                  "type Baz = { a: number };\n"
                                  "export function use(): Baz {\n  return { a: 1 };\n}\n"},
    })
    assert seen["shapes"] == ["sym:Baz,sym:use"]


def test_hook_names_the_shapes_being_written():
    """The feed's two inputs: every definition in the payload, or — for an Edit
    that changes a body, not a signature — the definition enclosing the edit,
    found by walking the target file upward from the edited lines."""
    src = HOOK.read_text()
    lib = (PLUGIN / "hooks" / "scribe_defs.sh").read_text()
    assert "scribe_defs()" in lib                      # one extractor, shared (#2901)
    assert "scribe_defs()" not in src                  # ...not a second copy here
    assert ".tool_input.old_string" in src             # the Edit's anchor
    assert "| tac | scribe_defs | head -1" in src      # nearest definition above
    # The ledger feed sends NAMES, never bodies, and stays on the one GET.
    assert src.count("/api/plugin/prior-art?") == 1


def _run_hook_against_sink(tmp_path, payload):
    """Run the hook with SCRIBE_URL pointed at a throwaway local listener and
    return the query the hook sent (tests.helpers.http_sink)."""
    with http_sink() as (port, seen):
        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}")
        out = subprocess.run(
            ["bash", str(HOOK)], input=json.dumps(payload),
            capture_output=True, text=True, env=env,
        )
        assert out.returncode == 0, out.stderr
    return seen[0] if seen else {}


def test_hook_sends_every_definition_in_a_write(tmp_path):
    repo = tmp_path / "repo"
    repo.mkdir()
    env = _hook_runtime_env()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
    seen = _run_hook_against_sink(tmp_path, {
        "session_id": "s-feed-w", "cwd": str(repo), "tool_name": "Write",
        "tool_input": {
            "file_path": str(repo / "new.ts"),
            "content": "export async function onDelete(): Promise<void> {\n"
                       "  const ok = await confirmed({ title: 'x' });\n}\n"
                       ".btn-primary {\n  color: red;\n}\n",
        },
    })
    assert seen["path"] == ["new.ts"]
    assert seen["shapes"] == ["css:btn-primary,sym:onDelete"]


def test_hook_sends_the_enclosing_definition_for_a_body_edit(tmp_path):
    """An Edit to the inside of onTrash names no definition itself; the hook
    must walk the file upward from the edited line and send onTrash."""
    import shutil
    if shutil.which("tac") is None:
        pytest.skip("the enclosing-definition walk needs tac")
    repo = tmp_path / "repo"
    repo.mkdir()
    env = _hook_runtime_env()
    subprocess.run(["git", "init", "-q"], cwd=repo, check=True, env=env)
    target = repo / "comp.vue"
    target.write_text(
        "<script setup lang=\"ts\">\n"
        "async function onTrash(): Promise<void> {\n"
        "  const ok = await confirmed({ title: 'Move to the trash?' });\n"
        "  if (!ok) return;\n"
        "}\n"
        "const other = () => {\n  return 1;\n};\n"
        "</script>\n"
    )
    seen = _run_hook_against_sink(tmp_path, {
        "session_id": "s-feed-e", "cwd": str(repo), "tool_name": "Edit",
        "tool_input": {
            "file_path": str(target),
            "old_string": "  if (!ok) return;",
            "new_string": "  if (!ok) return;\n  await guarded(() => store.trashNode(id));",
        },
    })
    assert seen["shapes"] == ["sym:onTrash"]


@pytest.mark.asyncio
async def test_the_write_time_derive_check_names_a_family_or_a_canon_in_band():
    """#2900: the ledger's own word on the names being written — a duplicate
    family with no canon, or a canon recorded elsewhere — rendered at the
    write even when nothing else does; keyed so the session's exclude
    channel silences a family already named."""
    from scribe.services import plugin_context as pc
    found = [
        {"symbol": "log-empty", "kind": "css", "key": "name:css:log-empty",
         "family": {"group": "name:css:log-empty", "label": ".log-empty", "identical": False,
                    "files": ["a/TaskLogSection.vue", "a/WorkspaceTaskPanel.vue"],
                    "file_count": 5, "size": 6,
                    "consumers": {"count": 6, "paths": ["a/TaskLogSection.vue", "a/V.vue"]}}},
        {"symbol": "btn-primary", "kind": "css", "key": "canon:2855",
         "canon": {"snippet_id": 2855, "path": "frontend/src/assets/components.css",
                   "label": ".btn-primary"}},
        {"symbol": "slugify", "kind": "sym", "key": "dup:483a",
         "family": {"group": "dup:483a", "label": "slugify", "identical": True,
                    "files": ["a/x.py", "a/y.py"], "file_count": 2, "size": 3}},
        {"symbol": "load", "kind": "sym", "key": "name:sym:load",
         "family": {"group": "name:sym:load", "label": "load", "identical": False,
                    "files": ["a/X.vue", "a/Y.vue", "a/Z.vue"], "file_count": 3, "size": 4}},
    ]
    check = AsyncMock(return_value=found)
    patches = dict(
        get_writepath_config=AsyncMock(return_value=_cfg()),
        semantic_search_notes=AsyncMock(return_value=[]),
        record_retrieval=MagicMock(), owner_names_for=AsyncMock(return_value={}),
    )
    with patch.multiple(pc, **patches), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \
         patch.object(pc.shape_ledger_svc, "write_time_derive", check):
        out = await pc.build_write_path_hint(
            1, "frontend/src/components/New.vue", code=REAL_CODE, project_id=24,
            stamp_shapes=[("css", "log-empty"), ("css", "btn-primary"), ("sym", "slugify"),
                          ("sym", "load")],
            exclude_derive=["name:sym:load"],
        )
    check.assert_awaited_once_with(24, "frontend/src/components/New.vue",
                                   [("css", "log-empty"), ("css", "btn-primary"),
                                    ("sym", "slugify"), ("sym", "load")])
    # The excluded family is gone; the other three render and are keyed.
    assert [d["key"] for d in out["derive"]] == ["name:css:log-empty", "canon:2855", "dup:483a"]
    assert out["derive_keys"] == ["name:css:log-empty", "canon:2855", "dup:483a"]
    ctx = out["context"]
    assert "Shape ledger at `frontend/src/components/New.vue`" in ctx
    # A CSS family is a repeated NAME (note 2917) and its dismissal is scoped-css;
    # a code dup family is an identical body and dismisses as convention-plumbing.
    # A css family says what renders it (milestone 302) before the ask.
    assert "`.log-empty` is a repeated name with no canon — defined in 5 other file(s): " \
           "`a/TaskLogSection.vue`, `a/WorkspaceTaskPanel.vue` +3 more; used by 6 templates: " \
           "`a/TaskLogSection.vue`, `a/V.vue` +4 more; derive it now" in ctx
    assert "`slugify` is a duplicate family with no canon — identical body in 2 other file(s): " \
           "`a/x.py`, `a/y.py`; derive it now" in ctx
    assert "`.btn-primary` is canon — snippet #2855 at `frontend/src/assets/components.css`" in ctx
    assert 'reason_code=\"scoped-css\")` dismisses' in ctx
    assert 'reason_code=\"convention-plumbing\")` dismisses' in ctx
    assert "`load`" not in ctx
    # No project → no check; a failing check never sinks the hint.
    check.reset_mock()
    with patch.multiple(pc, **patches), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \
         patch.object(pc.shape_ledger_svc, "write_time_derive", check):
        out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, stamp_shapes=[("sym", "f")])
    check.assert_not_awaited()
    assert out["derive"] == [] and out["derive_keys"] == []
    boom = AsyncMock(side_effect=RuntimeError("ledger down"))
    with patch.multiple(pc, **patches), \
         patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \
         patch.object(pc.shape_ledger_svc, "write_time_derive", boom):
        out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, project_id=24,
                                             stamp_shapes=[("sym", "f")])
    assert out["context"] == "" and out["derive"] == []


def test_the_hook_keeps_a_derive_channel_and_sends_it_back(tmp_path):
    """#2900: derive keys the server returns land in the session's own
    `.derive.ids` file and go back as `exclude_derive` on the next write —
    a family is named once per session, not at every edit."""
    reply = (b'{"context":"> family","note_ids":[],"sync_note_ids":[],'
             b'"derive_keys":["dup:483a","canon:2855"]}')
    with http_sink(reply) as (port, seen):
        env = dict(_hook_runtime_env(), SCRIBE_URL=f"http://127.0.0.1:{port}",
                   TMPDIR=str(tmp_path))
        payload = {"session_id": "s-derive-1", "cwd": str(tmp_path), "tool_name": "Write",
                   "tool_input": {"file_path": str(tmp_path / "a.css"),
                                  "content": ".log-empty {\n  color: red;\n}\n"}}
        for _ in range(2):
            out = subprocess.run(["bash", str(HOOK)], input=json.dumps(payload),
                                 capture_output=True, text=True, env=env)
            assert out.returncode == 0, out.stderr
    assert "exclude_derive" not in seen[0]
    assert seen[1]["exclude_derive"] == ["dup:483a,canon:2855"]
    state = tmp_path / "scribe-priorart" / "s-derive-1.derive.ids"
    assert state.read_text().split() == ["dup:483a", "canon:2855", "dup:483a", "canon:2855"]


@pytest.mark.asyncio
async def test_the_write_time_divergence_check_is_named_in_band():
    """#2793: the hook named a shape at a path whose directory a canon
    dominates, and the stamp didn't make it that canon's instance — the hint
    must say so at the write, even when nothing else renders."""
    from scribe.services import plugin_context as pc
    div = [{"symbol": "confirmDanger", "kind": "sym", "canon_snippet_id": 2761,
            "instances": 20, "judged": 21}]
    check = AsyncMock(return_value=div)
    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()), \
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "write_time_divergence", check):
        out = await pc.build_write_path_hint(
            1, "frontend/src/components/Danger.vue", code=REAL_CODE, project_id=24,
            stamp_shapes=[("sym", "confirmDanger")],
        )
    check.assert_awaited_once_with(24, "frontend/src/components/Danger.vue",
                                   [("sym", "confirmDanger")], [])
    assert out["divergence"] == div
    assert "Divergence check at `frontend/src/components/Danger.vue`" in out["context"]
    assert "`confirmDanger` → #2761 (20 of 21 judged siblings are its instances)" in out["context"]
    assert "variant" in out["context"]
    # No project → no check at all.
    check.reset_mock()
    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()), \
         patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
         patch.object(pc.shape_ledger_svc, "write_time_divergence", check):
        out = await pc.build_write_path_hint(1, "x.py", code=REAL_CODE, stamp_shapes=[("sym", "f")])
    check.assert_not_awaited()
    assert out["divergence"] == []


def test_hook_keeps_the_rule_channel_apart_from_the_other_three():
    """Milestone 307's arm, pinned the way #2708's was.

    Standing rules dedup on their OWN file and their OWN query parameter. One
    shared channel is the bug #2708 already fixed once: a hint of one class
    silencing a different class that had never been shown. A rule named early
    must not be re-offered on every later write, and must not silence — or be
    silenced by — a snippet suggestion.
    """
    src = HOOK.read_text()
    assert ".rules.ids" in src                      # its own state file
    assert "exclude_rule_ids=" in src               # its own query channel
    assert "(.rule_ids // [])[]?" in src            # its own write-back
    # And it rides the same request as the rest, not a second round trip.
    assert "${rule_exclude_q}" in src