FabledScribe/tests/test_guidance_ownership.py

"""Every piece of agent guidance has one owner — and none of it is lost on the way there.

WHY THIS EXISTS (milestone 410, decision #4027)

Scribe's guidance to agents was written up to five times over: the MCP
`_INSTRUCTIONS`, the plugin's static session context, the live session context
the server builds, the `using-scribe` skill, and tool docstrings. An earlier
design (#2494) made that deliberate — insurance against any one surface failing
silently — and the copies drifted apart instead (#2497, #4022).

Decision #4027 replaced the redundancy with ownership: the server orients, the
skills hold the depth, and each client adapter holds only its own timing and
conventions. This module is the registry that decision is enforced from.

WHAT IT PINS NOW, AND WHAT COMES LATER

Step 1 — the LOSS GUARD. Every topic below must still be stated on at least one
surface a session actually receives. Milestone 410's later steps delete copies;
this is what stops the last copy of a topic going with them.

Step 6 will add the OWNERSHIP guard: a topic's full statement on its `owner`
and nowhere else. The `owner` column is recorded here from the start so the
registry is written once, but nothing asserts it yet — during the moves a
topic is legitimately in several places at once.

MARKERS ARE PHRASES, NOT WORDS

A topic is "stated" when ALL of its markers appear on one surface — so a
topic's markers must travel together, and a lone common word never counts
(BINDING_CLAIMS in test_instruction_surfaces_agree explains the false alarm a
bare word raises). Tool names are the preferred marker: they change only when
the tool does. Reword a topic deliberately and update its markers in the same
commit; the failure message names which marker went missing where.
"""
from __future__ import annotations

import pathlib
import re
from typing import NamedTuple

ROOT = pathlib.Path(__file__).resolve().parents[1]


def _norm(text: str) -> str:
    # Whitespace-flattened and lowercased: prose is hard-wrapped, so a marker
    # can straddle a line break without the guidance having changed.
    return " ".join(text.split()).lower()


def _live_session_context_source() -> str:
    """The source of build_session_context — the prose it emits lives there.

    Read as text rather than called: the function needs a database, and what
    this module checks is what the product SAYS, which is the literal strings.
    """
    src = (ROOT / "src/scribe/services/plugin_context.py").read_text()
    start = src.index("async def build_session_context")
    nxt = re.search(r"\n(?:async def|def) ", src[start + 1:])
    return src[start:start + 1 + nxt.start()] if nxt else src[start:]


def delivered_surfaces() -> dict[str, str]:
    """Every surface a session receives as guidance, by label.

    The one definition of "delivered" for this module and its successors:
      - `instructions` — the MCP server's `_INSTRUCTIONS` (every MCP client)
      - `docstrings`   — the MCP tool modules (tool descriptions, every client)
      - `skill:<name>` — each bundled Agent Skill
      - `static`       — the Claude Code adapter's static session context
      - `commands`     — the Claude Code adapter's slash commands
      - `live`         — the live session context the server builds
    """
    server = (ROOT / "src/scribe/mcp/server.py").read_text()
    match = re.search(r'_INSTRUCTIONS = """(.*?)"""', server, re.S)
    assert match, "server.py no longer defines _INSTRUCTIONS as a triple-quoted literal"
    surfaces = {
        "instructions": match.group(1),
        "docstrings": "".join(p.read_text() for p in sorted((ROOT / "src/scribe/mcp/tools").glob("*.py"))),
        "static": (ROOT / "plugin/hooks/scribe_static_context.md").read_text(),
        "commands": "".join(p.read_text() for p in sorted((ROOT / "plugin/commands").glob("*.md"))),
        "live": _live_session_context_source(),
    }
    for skill in sorted((ROOT / "plugin/skills").glob("*/SKILL.md")):
        surfaces[f"skill:{skill.parent.name}"] = skill.read_text()
    return {label: _norm(text) for label, text in surfaces.items()}


class Topic(NamedTuple):
    key: str
    owner: str  # a label from delivered_surfaces(); asserted from step 6
    markers: tuple[str, ...]


# The ownership map from milestone 410's body, one row per topic, plus the
# topics once guarded as "displaced from _INSTRUCTIONS" (#2562), folded in so
# there is one list. Retired topics (the surface-precedence tiebreaker) are
# absent on purpose: nothing has to keep saying them.
TOPICS: tuple[Topic, ...] = (
    # ── the working reflexes — owned by the using-scribe skill ──
    Topic("scribe is the system of record; keep one copy", "skill:using-scribe", ("one copy",)),
    Topic("orient: enter the project, check repo bindings", "skill:using-scribe",
          ("enter_project", "list_repo_bindings")),
    Topic("rules are retrieved; ask before a consequential act", "skill:using-scribe",
          ('content_type="rule"', "nothing matched")),
    Topic("rules bind, preferences guide and are kept current", "skill:using-scribe",
          ("preference", "update_preference")),
    Topic("recall before acting", "skill:using-scribe", ("recall before acting",)),
    Topic("stay inside the active project's scope", "skill:using-scribe",
          ("stay inside the active project", "cross-project")),
    Topic("record as you go; honest status; fixes are issues", "skill:using-scribe",
          ("add_task_log", "in_progress", 'kind="issue"')),
    Topic("an id exists only once a create returns it", "skill:using-scribe",
          ("exists only once a create", "{{ref:")),
    Topic("tag records to systems as you write", "skill:using-scribe", ("system_ids", "create_system")),
    Topic("answer the systems_hint at the moment of work", "skill:using-scribe", ("systems_hint",)),
    Topic("a retrieved rule outranks a default habit", "skill:using-scribe", ("outranks a default habit",)),
    Topic("log on completion and on a problem", "skill:using-scribe", ("hit or discover a problem",)),
    Topic("the project's design system binds ui", "skill:using-scribe", ("resolve_design_system",)),
    Topic("name the record, never just its number", "skill:using-scribe", ("name the record",)),
    Topic("project inception is a decision", "skill:using-scribe", ("decide_project_inception",)),
    Topic("where a new rule goes, and its trigger", "skill:using-scribe",
          ("create_project_rule", "when_to_apply")),
    Topic("a rule vs the other entities", "skill:using-scribe", ("standing instruction",)),
    Topic("reference notes update in place; dev-logs don't", "skill:using-scribe", ("reference note",)),
    # ── process arcs — owned by their skills ──
    Topic("plan in a milestone, steps created together", "skill:writing-plans", ("start_planning", "{{ref:")),
    Topic("reuse recorded shapes; record at first build", "skill:reusing-code",
          ("create_snippet", "when_to_use", "first build", "second copy")),
    Topic("report back where the work stands", "skill:reporting-back", ("reporting-back", "placement")),
    # ── per-tool contracts and in-band behaviour — owned by the server ──
    Topic("closing a task cues the report", "docstrings", ("report_back",)),
    Topic("a note that asserts a fact carries its check", "docstrings", ("verify_with", "expires_when")),
    Topic("supersession demotes, never hides", "docstrings", ("supersedes",)),
    Topic("deletes are recoverable from the trash", "docstrings", ("deleted_batch_id",)),
    Topic("creates are duplicate-gated", "docstrings", ("near-duplicate",)),
    Topic("shared records are another user's suggestion", "docstrings", ("shared: true",)),
    Topic("stored processes are followed verbatim", "docstrings", ("stored processes", "verbatim")),
    Topic("a project is never guessed", "docstrings", ("never guessing a project",)),
    Topic("an unbound repo gets a bind hint", "live", ("bind_repo",)),
    # ── the Claude Code adapter's own conventions ──
    Topic("compact at clean seams", "static", ("/compact",)),
    Topic("stored processes sync into local skills", "commands", ("scribe-proc-",)),
    Topic("say so when scribe's tools are unavailable", "static", ("tools are unavailable",)),
)


def missing_topics(topics, surfaces: dict[str, str]) -> list[str]:
    """Topics no single surface states in full — with the nearest miss named.

    Pure, so the guard's ability to fail is itself testable.
    """
    missing = []
    for topic in topics:
        markers = [m.lower() for m in topic.markers]
        if any(all(m in text for m in markers) for text in surfaces.values()):
            continue
        partial = {
            label: [m for m in markers if m not in text]
            for label, text in surfaces.items()
            if any(m in text for m in markers)
        }
        missing.append(f"{topic.key!r} — markers {topic.markers}; nearest: {partial or 'nowhere'}")
    return missing


def test_no_guidance_topic_has_fallen_off_every_surface():
    missing = missing_topics(TOPICS, delivered_surfaces())
    assert not missing, (
        "these guidance topics are no longer stated in full on ANY delivered "
        "surface:\n  " + "\n  ".join(missing) + "\nMilestone 410 moves guidance "
        "to one owner per topic (decision #4027); a move that deletes a copy "
        "must leave the topic stated on its owner. If the topic was reworded on "
        "purpose, update its markers here in the same commit."
    )


def test_every_owner_is_a_surface_that_exists():
    labels = set(delivered_surfaces())
    unknown = [(t.key, t.owner) for t in TOPICS if t.owner not in labels]
    assert not unknown, f"owners that name no delivered surface: {unknown}"


def test_topic_keys_are_unique():
    keys = [t.key for t in TOPICS]
    assert len(keys) == len(set(keys))


def test_the_loss_guard_can_fail():
    """Rule 167: a guard that cannot fail protects nothing.

    A topic whose markers are split across two surfaces is NOT stated — the
    phrases have to travel together — and one whose marker is nowhere is
    reported with 'nowhere'.
    """
    surfaces = {"a": "enter_project here", "b": "list_repo_bindings there"}
    split = Topic("split", "a", ("enter_project", "list_repo_bindings"))
    absent = Topic("absent", "a", ("no such phrase",))
    whole = Topic("whole", "a", ("enter_project",))
    reported = missing_topics((split, absent, whole), surfaces)
    assert len(reported) == 2
    assert reported[0].startswith("'split'") and "nowhere" in reported[1]


# ── The skills are client-neutral (milestone 410 step 2) ────────────────
#
# Agent Skills is an open format read by dozens of clients (#4023), so the
# skills folder is the part of every client package that is shared verbatim.
# Anything only one client understands belongs in that client's adapter —
# for Claude Code, the plugin's static context, hooks and commands — or the
# skill reads as nonsense everywhere else. Each marker below names a thing
# exactly one client has:
#   - "claude", "~/.claude", ".claude/"   the client itself and its paths
#   - "claude.md", "memory.md", "auto-memory", "automemoryenabled"
#                                          Claude Code's local memory files/setting
#   - "/compact", "/scribe:"               Claude Code slash commands
#   - "sessionstart", "userpromptsubmit", "pretooluse", "posttooluse"
#                                          Claude Code hook event names
#   - "write/edit"                         Claude Code's editor tool names
#   - "claude_plugin_root"                 the Claude Code plugin root variable
# `Bash` is matched case-sensitively as a word: it is Claude Code's shell tool
# name, while "bash" in prose is just the shell.
CLIENT_SPECIFIC = (
    "claude", "~/.claude", ".claude/", "claude.md", "memory.md", "auto-memory",
    "automemoryenabled", "/compact", "/scribe:", "sessionstart", "userpromptsubmit",
    "pretooluse", "posttooluse", "write/edit", "claude_plugin_root",
)
CLIENT_TOOL_NAME = re.compile(r"\bBash\b")


def client_specific_hits(text: str) -> list[str]:
    hits = [m for m in CLIENT_SPECIFIC if m in text.lower()]
    if CLIENT_TOOL_NAME.search(text):
        hits.append("Bash")
    return hits


def test_the_skills_name_no_particular_client():
    offenders = {
        str(p.relative_to(ROOT)): client_specific_hits(p.read_text())
        for p in sorted((ROOT / "plugin/skills").glob("*/SKILL.md"))
    }
    offenders = {path: hits for path, hits in offenders.items() if hits}
    assert not offenders, (
        f"skills that name one client: {offenders}. Skills are shared by every "
        f"client package (decision #4027); say the universal thing in the skill "
        f"and put the client's own name for it in that client's adapter."
    )


def test_the_client_guard_can_fail():
    assert client_specific_hits("keep a copy in CLAUDE.md") == ["claude", "claude.md"]
    assert client_specific_hits("edits made through Bash") == ["Bash"]
    assert client_specific_hits("edits made through a bash shell") == []