FabledScribe/tests/test_derived_mirror_generic_door.py

"""A DERIVED `data` mirror survives the GENERIC note door.

Some kinds store a queryable mirror in `notes.data` that is derived from the
body: a snippet (name, when_to_use, locations…) and a lesson (what, the
trigger, what taught it). Their own updaters compose it from the field set they
just merged, so those were never the problem — the problem is every other way
the body can be written. `update_note` is a `hasattr` loop, and both doors
reach it: PATCH /api/notes/<id> and the MCP update_note tool. The Knowledge
feed hands you that path, because a card there routes to /notes/:id.

The failure is silent and the wrong way round, because every read PREFERS the
mirror — deliberately, since parsing markdown to answer what an index can
answer is how a hot path rots.

  - A SNIPPET went on reporting its old repo/path/symbol to the location
    reverse lookup and to prior-art recall while displaying its new body: a
    record surfaced with full authority and wrong, which the drift-check
    docstring calls worse than having no record at all (#3128).

  - A LESSON is worse. Its mirror holds the TRIGGER, and the trigger is the
    entire retrieval story — a stale one keeps the lesson firing for the
    situation it used to name while it displays the one it now names.

#3128's fix was correct and did not generalise: it tested one constant, so
milestone 385's lesson arrived with the same design and none of the protection.
The registry tests below assert the PROPERTY — every kind with a derived mirror
is registered — so a third kind fails here rather than shipping quiet (#3734).
"""
import inspect

import pytest
from tests.helpers import drive_update_note as _update
from tests.helpers import fake_lesson, fake_note, fake_snippet

OLD_MIRROR = {
    "name": "debounce",
    "language": "javascript",
    "locations": [{"repo": "Scribe", "path": "old/place.js", "symbol": "debounce"}],
    "verification": {"status": "ok", "code_sha": "abc", "checked_at": "2026-01-01"},
    "provenance": {"commit_sha": "deadbeef"},
}

MOVED_BODY = (
    "**Locations:**\n"
    "- `Scribe` · `new/place.ts` · `debounce`\n\n"
    "```typescript\nexport const debounce = 1;\n```\n"
)


@pytest.mark.asyncio
async def test_a_body_write_moves_the_mirror_with_it():
    note = fake_snippet(data=dict(OLD_MIRROR), project_id=None)
    await _update(note, body=MOVED_BODY)
    assert note.data["locations"] == [
        {"repo": "Scribe", "path": "new/place.ts", "symbol": "debounce"}
    ], "the mirror still describes where the snippet used to live"
    assert note.data["language"] == "typescript"


@pytest.mark.asyncio
async def test_the_verdict_and_provenance_are_carried_not_dropped():
    """Neither is in the body to parse, so recomposing must carry them. An
    ordinary edit must not erase the last drift check — and it needs no
    invalidation branch either: `code_sha` is recomputed from the new code, so
    a verdict stamped against the old code expires itself on read."""
    note = fake_snippet(data=dict(OLD_MIRROR), project_id=None)
    await _update(note, body=MOVED_BODY)
    assert note.data["verification"] == OLD_MIRROR["verification"]
    assert note.data["provenance"] == OLD_MIRROR["provenance"]
    assert note.data["code_sha"] != OLD_MIRROR["verification"]["code_sha"]


@pytest.mark.asyncio
async def test_an_explicit_data_wins_over_recomposition():
    """`update_snippet` composes the mirror from the merged field set it holds
    and passes it here. That caller knows things the body cannot be re-read for
    — which locations were replaced, whether provenance survives the edit — so
    an explicit mirror must not be recomputed out from under it."""
    note = fake_snippet(data=dict(OLD_MIRROR), project_id=None)
    authoritative = {"name": "from the service", "locations": []}
    await _update(note, body=MOVED_BODY, data=authoritative)
    assert note.data == authoritative


@pytest.mark.asyncio
async def test_a_plain_note_is_left_alone():
    """Only snippets carry a mirror; a note's `data` must not be invented."""
    note = fake_note(note_type="note", data=None, project_id=None)
    await _update(note, body="just some prose")
    assert note.data is None


@pytest.mark.asyncio
async def test_a_write_that_cannot_change_the_parse_does_not_touch_the_mirror():
    """Status, priority, project — none of them is an input to the body parser,
    so recomposing on them would be work for nothing and would rebuild a mirror
    from a body nobody claimed to have changed."""
    note = fake_snippet(data=dict(OLD_MIRROR), project_id=None)
    await _update(note, project_id=4)
    assert note.data == OLD_MIRROR


@pytest.mark.asyncio
async def test_a_title_change_reaches_the_mirror_too():
    """A snippet's NAME lives in its title, not its body — `parse_snippet_fields`
    reads both, so both are triggers."""
    note = fake_snippet(data=dict(OLD_MIRROR), project_id=None)
    await _update(note, title="throttle — cap a callback's rate")
    assert note.data["name"] == "throttle"
    assert note.data["when_to_use"] == "cap a callback's rate"


# ── the registry, rather than a chain of ifs ─────────────────────────────────


def test_every_kind_with_a_derived_mirror_is_registered():
    """The load-bearing one. Before #3734 this was a single `if` naming
    snippets, and the kind added next was simply not covered."""
    from scribe.services.lessons import LESSON_NOTE_TYPE
    from scribe.services.notes import _mirror_recomposers
    from scribe.services.snippets import SNIPPET_NOTE_TYPE

    table = _mirror_recomposers()
    assert SNIPPET_NOTE_TYPE in table, "the #3128 fix was lost"
    assert LESSON_NOTE_TYPE in table, (
        "a lesson's `data` holds its trigger and `lesson_trigger` prefers it, "
        "so a body edit through the generic door would leave the lesson being "
        "retrieved for a situation it no longer names (#3734)"
    )


def test_the_dispatch_is_a_lookup_not_a_named_kind():
    """Asserted on structure (rule 167). A lookup extends in one line in one
    place; naming a kind inline is the shape that left lessons uncovered."""
    from scribe.services import notes as notes_module

    src = inspect.getsource(notes_module.update_note)
    assert "_mirror_recomposers()" in src
    assert "SNIPPET_NOTE_TYPE" not in src, (
        "update_note names one kind again — that is the shape #3734 replaced"
    )


def test_each_recomposer_takes_the_note_and_nothing_else():
    """A registry entry with the wrong signature fails inside a generic PATCH,
    which is the one moment nobody is watching."""
    from scribe.services.notes import _mirror_recomposers

    for kind, fn in _mirror_recomposers().items():
        assert callable(fn), f"{kind} maps to something not callable"
        assert len(inspect.signature(fn).parameters) == 1, kind


# ── a lesson's mirror moves with its body ────────────────────────────────────


NEW_TRIGGER = "a CI run has sat in_progress far longer than its suite takes"


def _lesson_body(trigger, insight="Read the job log.", sources=None):
    from scribe.services.lessons import compose_body
    return compose_body(insight, trigger, sources)


@pytest.mark.asyncio
async def test_a_body_write_moves_a_lessons_trigger_with_it():
    from scribe.services.lessons import TRIGGER_KEY, compose_title

    what = "Read the job log before waiting longer"
    note = fake_lesson(
        title=compose_title(what, NEW_TRIGGER),
        data={TRIGGER_KEY: "a CI run is slow", "what": what},
        project_id=None,
    )
    await _update(note, body=_lesson_body(NEW_TRIGGER))
    assert note.data[TRIGGER_KEY] == NEW_TRIGGER, (
        "the mirror kept the old trigger — the lesson would still be retrieved "
        "for the situation it no longer names"
    )
    assert note.data["what"] == what


@pytest.mark.asyncio
async def test_a_subject_containing_an_em_dash_still_splits():
    """Why `untrigger_title` is given the trigger instead of splitting on the
    separator: a subject may legitimately contain one."""
    from scribe.services.lessons import TRIGGER_KEY, compose_title

    what = "A wait with no deadline — the shape, not the symptom"
    trigger = "you are about to await something crossing a process boundary"
    note = fake_lesson(
        title=compose_title(what, trigger), data=None, project_id=None,
    )
    await _update(note, body=_lesson_body(trigger))
    assert note.data["what"] == what
    assert note.data[TRIGGER_KEY] == trigger


@pytest.mark.asyncio
async def test_dropping_the_provenance_line_drops_it_from_the_mirror():
    """The body is the authority. Carrying a value the reader just deleted is
    the failure this recompose exists to prevent, not a courtesy — the
    opposite call from a snippet's `verification`, which is carried because it
    was never in the body to delete."""
    from scribe.services.lessons import SOURCES_KEY, compose_title

    note = fake_lesson(
        title=compose_title("Something learned", "a situation"),
        data={SOURCES_KEY: [999]},
        project_id=None,
    )
    await _update(note, body=_lesson_body("a situation"))   # no Learned from:
    assert SOURCES_KEY not in note.data


@pytest.mark.asyncio
async def test_an_explicit_data_wins_for_a_lesson_too():
    """`update_lesson` composes the mirror from the merged field set and passes
    it here; that caller knows things a re-read of the body cannot recover."""
    from scribe.services.lessons import TRIGGER_KEY

    note = fake_lesson(data={TRIGGER_KEY: "old"}, project_id=None)
    authoritative = {TRIGGER_KEY: "from the service", "what": "x"}
    await _update(note, body=_lesson_body(NEW_TRIGGER), data=authoritative)
    assert note.data == authoritative


@pytest.mark.asyncio
async def test_a_lesson_title_change_reaches_the_mirror():
    """A lesson's subject lives in its title, so a title edit is a trigger for
    recomposition exactly as it is for a snippet's name."""
    from scribe.services.lessons import TRIGGER_KEY, compose_title

    trigger = "two absolute siblings overlap"
    note = fake_lesson(
        body=_lesson_body(trigger),
        data={TRIGGER_KEY: trigger, "what": "the old subject"},
        project_id=None,
    )
    await _update(note, title=compose_title("the new subject", trigger))
    assert note.data["what"] == "the new subject"
    assert note.data[TRIGGER_KEY] == trigger


# ── the join and its inverse ─────────────────────────────────────────────────


@pytest.mark.parametrize(
    ("subject", "trigger"),
    [
        ("a subject", "a trigger"),
        ("a subject — with a dash", "a trigger"),
        ("a subject", ""),
        ("", "a trigger"),
        ("a subject", "a trigger — with a dash"),
    ],
)
def test_untrigger_title_inverts_trigger_title(subject, trigger):
    """#3207's shape: the join had three copies before it was consolidated, so
    its inverse lives beside it rather than in whichever caller wanted it."""
    from scribe.services.embeddings import trigger_title, untrigger_title

    title = trigger_title(subject, trigger)
    assert untrigger_title(title, trigger) == (subject or trigger).strip()


def test_untrigger_title_degrades_to_the_whole_title():
    """A record written before the join existed still answers with something a
    human recognises rather than with ""."""
    from scribe.services.embeddings import untrigger_title

    assert untrigger_title("a plain old title", "") == "a plain old title"
    assert untrigger_title("a plain old title", "a trigger it lacks") == (
        "a plain old title"
    )


def test_the_trigger_separator_is_spelled_in_exactly_one_place():
    """Two literals that must match are one edit away from not matching."""
    import pathlib

    root = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
    offenders = [
        str(p.relative_to(root)) for p in root.rglob("*.py")
        if '" \u2014 "' in p.read_text() and p.name != "embeddings.py"
    ]
    assert not offenders, (
        f"the trigger separator is spelled inline in {offenders} — use "
        f"TRIGGER_SEP, trigger_title or untrigger_title (#3207)"
    )


def test_the_mirror_guards_can_fail():
    """Rule 167: shown turning red once."""
    from scribe.services.lessons import TRIGGER_KEY, recompose_data

    bare = fake_lesson(title="just a title", body="no composed lines", data=None)
    assert TRIGGER_KEY not in recompose_data(bare)