FabledScribe/tests/test_guidance_ownership.py

"""Every piece of agent guidance has exactly one owner that states it.

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 with
a short index, the skills hold the depth, tool docstrings hold each tool's
contract, and each client adapter holds only its own timing and conventions.
This module is the registry that decision is enforced from.

WHAT IT PINS

1. OWNERS STATE THEIR TOPIC. Every topic's markers and its distinctive
   statement appear on its owner — so moving or trimming text cannot drop a
   topic without failing here.
2. NOWHERE ELSE STATES IT IN FULL. The statement — a phrase distinctive to
   the owner's full wording — must not appear on any other session surface
   (the index, the adapter's static text and commands, the live context, the
   other skills). A topic legitimately stated in two places at two different
   moments declares that in `shared_with`, with the reason beside it.
3. THE INDEX NAMES THE SESSION-START REFLEXES. `_INSTRUCTIONS` is the one
   surface every MCP client receives, so each reflex it indexes keeps its
   `index` markers there — a one-line pointer, not a copy.

WHAT IT CANNOT SEE

A copy reworded so it no longer contains the statement phrase passes. The
guard catches the ordinary way duplication happens — pasting a paragraph
into a second surface — not a determined paraphrase. Tool docstrings are not
scanned for copies: a tool's contract may elaborate the reflex that calls it.

MARKERS AND STATEMENTS ARE PHRASES, NOT WORDS

Reword a topic deliberately and update its markers/statement in the same
commit; each failure message names the topic, the surface and the phrase.
"""
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 phrase
    # 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:
      - `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()}


# Not scanned for copies (see the module docstring).
NOT_COPY_SCANNED = frozenset({"docstrings"})


class Topic(NamedTuple):
    key: str
    owner: str                     # a label from delivered_surfaces()
    markers: tuple[str, ...]       # what the topic is about; all on the owner
    statement: str                 # distinctive to the owner's full wording
    index: tuple[str, ...] = ()    # required in _INSTRUCTIONS, if it indexes this
    shared_with: tuple[str, ...] = ()  # other surfaces allowed the statement, with a reason


U = "skill:using-scribe"

TOPICS: tuple[Topic, ...] = (
    # ── the working reflexes — owned by the using-scribe skill ──
    Topic("scribe is the system of record; keep one copy", U, ("one copy",),
          "let any existing local memory shrink", index=("one copy",)),
    Topic("orient: enter the project, check repo bindings", U, ("enter_project", "list_repo_bindings"),
          "returns the project plus the rules bound to the areas it works in",
          index=("enter_project",)),
    Topic("rules are retrieved; ask before a consequential act", U, ('content_type="rule"', "nothing matched"),
          "an empty session is not evidence of an empty rulebook",
          index=('content_type="rule"', "nothing matched")),
    Topic("rules bind, preferences guide and are kept current", U, ("preference", "update_preference"),
          "a preference is the one record you keep current yourself", index=("preferences guide",)),
    Topic("recall before acting", U, ("recall before acting",), "for related prior work", index=("recall",)),
    Topic("stay inside the active project's scope", U, ("stay inside the active project", "cross-project"),
          "stay inside the active project's scope", index=("project_id",)),
    Topic("record as you go; honest status; fixes are issues", U, ("add_task_log", "in_progress", 'kind="issue"'),
          "fixes are issues, not work-logs", index=("add_task_log", 'kind="issue"')),
    Topic("an id exists only once a create returns it", U, ("exists only once a create", "{{ref:"),
          "exists only once a create call returns it", index=("create_records", "{{ref:n}}")),
    Topic("tag records to systems as you write", U, ("system_ids", "create_system"),
          "would someone investigating that subsystem want this record", index=("system_ids",)),
    Topic("answer the systems_hint at the moment of work", U, ("systems_hint",),
          "treat it as the tagging question asked at the moment of work"),
    Topic("a retrieved rule outranks a default habit", U, ("outranks a default habit",),
          "a retrieved rule outranks a default habit"),
    Topic("log on completion and on a problem", U, ("hit or discover a problem",), "hit or discover a problem"),
    Topic("the project's design system binds ui", U, ("resolve_design_system",),
          "building ui: the project's design system binds", index=("resolve_design_system",)),
    Topic("name the record, never just its number", U, ("name the record",),
          "a bare id reads as complete to you and as homework to them"),
    Topic("project inception is a decision", U, ("decide_project_inception",),
          "starting a project: decide what it inherits", index=("decide_project_inception",)),
    Topic("where a new rule goes, and its trigger", U, ("create_project_rule", "when_to_apply"),
          "whichever home it gets"),
    Topic("a rule vs the other entities", U, ("standing instruction",), "first ask whether it's a rule at all"),
    Topic("reference notes update in place; dev-logs don't", U, ("reference note",),
          "state updates in place; chronicles don't"),
    # ── process arcs — owned by their skills ──
    Topic("plan in a milestone, steps created together", "skill:writing-plans", ("start_planning", "{{ref:"),
          "a milestone earns its place when the work has an arc", index=("start_planning",)),
    Topic("reuse recorded shapes; record at first build", "skill:reusing-code",
          ("create_snippet", "when_to_use", "first build", "second copy"),
          "prior art offered beside a write is not noise", index=("create_snippet",)),
    Topic("report back where the work stands", "skill:reporting-back", ("reporting-back", "placement"),
          "take the placement from the record", index=("placement",)),
    # ── per-tool contracts and in-band behaviour — owned by the server ──
    Topic("closing a task cues the report", "docstrings", ("report_back",), "reporting this to the operator?"),
    Topic("a note that asserts a fact carries its check", "docstrings", ("verify_with", "expires_when"),
          "could this note become false without anyone editing it",
          # Stated at two different moments on purpose: the tool contract is
          # read when the field is about to be filled, the skill while deciding
          # what to write at all (test_verification_guidance_survives pins both).
          shared_with=(U,)),
    Topic("supersession demotes, never hides", "docstrings", ("supersedes",),
          "it simply stops competing with this one"),
    Topic("deletes are recoverable from the trash", "docstrings", ("deleted_batch_id",), "deleted_batch_id"),
    Topic("creates are duplicate-gated", "docstrings", ("near-duplicate",), "bypass the near-duplicate gate",
          index=("duplicate-gated",)),
    Topic("shared records are another user's suggestion", "docstrings", ("shared: true",),
          "belongs to another user", index=("shared:true",)),
    Topic("stored processes are followed as written", "docstrings", ("get_process",), "follow the returned body"),
    Topic("a project is never guessed", "docstrings", ("never guessing a project",), "never guessing a project"),
    Topic("an unbound repo gets a bind hint", "live", ("bind_repo",), "isn't mapped to a scribe project"),
    # ── the Claude Code adapter's own conventions ──
    Topic("compact at clean seams", "static", ("/compact",), "compact at clean seams"),
    Topic("stored processes sync into local skills", "commands", ("scribe-proc-",), "regenerate the local skill stubs"),
    Topic("say so when scribe's tools are unavailable", "static", ("tools are unavailable",),
          "rather than silently falling back to local notes"),
)


def owner_gaps(topics, surfaces: dict[str, str]) -> list[str]:
    """Topics whose owner no longer carries every marker and the statement."""
    gaps = []
    for t in topics:
        text = surfaces.get(t.owner, "")
        absent = [p for p in (*t.markers, t.statement) if p.lower() not in text]
        if absent:
            elsewhere = [label for label, other in surfaces.items() if t.statement.lower() in other]
            gaps.append(f"{t.key!r} on {t.owner}: missing {absent}; statement found on {elsewhere or 'nothing'}")
    return gaps


def copies(topics, surfaces: dict[str, str]) -> list[str]:
    """Topics whose full statement also appears on a surface that doesn't own it."""
    found = []
    for t in topics:
        allowed = {t.owner, *t.shared_with}
        extra = [label for label, text in surfaces.items()
                 if label not in NOT_COPY_SCANNED and label not in allowed and t.statement.lower() in text]
        if extra:
            found.append(f"{t.key!r} (owner {t.owner}) is also stated on {extra}")
    return found


def index_gaps(topics, surfaces: dict[str, str]) -> list[str]:
    text = surfaces.get("instructions", "")
    return [f"{t.key!r}: {[m for m in t.index if m.lower() not in text]}"
            for t in topics if any(m.lower() not in text for m in t.index)]


def test_every_topic_is_stated_by_its_owner():
    gaps = owner_gaps(TOPICS, delivered_surfaces())
    assert not gaps, (
        "these topics are no longer stated in full by their owner:\n  "
        + "\n  ".join(gaps) + "\nEvery guidance topic has one owner (decision "
        "#4027). If the text moved on purpose, move the topic's owner here in the "
        "same commit; if it was reworded, update its markers and statement."
    )


def test_no_topic_is_stated_in_full_off_its_owner():
    found = copies(TOPICS, delivered_surfaces())
    assert not found, (
        "guidance stated in full on a surface that doesn't own it:\n  "
        + "\n  ".join(found) + "\nOne owner per topic (decision #4027): replace "
        "the copy with a one-line pointer to the owner. If both places genuinely "
        "need it at different moments, declare it in `shared_with` with the reason."
    )


def test_the_index_names_each_reflex_it_points_at():
    gaps = index_gaps(TOPICS, delivered_surfaces())
    assert not gaps, (
        f"_INSTRUCTIONS no longer carries the index line for: {gaps}. It is the "
        f"one guidance surface every MCP client receives; keep a one-line pointer "
        f"per session-start reflex (see the comment above _INSTRUCTIONS)."
    )


def test_every_owner_and_sharer_is_a_surface_that_exists():
    labels = set(delivered_surfaces())
    unknown = [(t.key, s) for t in TOPICS for s in (t.owner, *t.shared_with) if s not in labels]
    assert not unknown, f"owners or sharers 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_ownership_guards_can_fail():
    """Rule 167: each guard is shown turning red once."""
    surfaces = {
        "instructions": "use the widget",
        "docstrings": "the widget owner statement lives here",
        "skill:a": "widget owner statement lives here, and widget tool",
        "skill:b": "a pasted copy: widget owner statement lives here",
        "static": "",
    }
    topic = Topic("widget", "skill:a", ("widget tool",), "widget owner statement lives here",
                  index=("use the widget", "not in the index"))
    moved = topic._replace(owner="static")
    assert owner_gaps((topic,), surfaces) == []
    assert owner_gaps((moved,), surfaces) and "missing" in owner_gaps((moved,), surfaces)[0]
    # skill:b pasted it; docstrings are not scanned for copies.
    assert copies((topic,), surfaces) == ["'widget' (owner skill:a) is also stated on ['skill:b']"]
    assert copies((topic._replace(shared_with=("skill:b",)),), surfaces) == []
    assert index_gaps((topic,), surfaces) == ["'widget': ['not in the index']"]


# ── 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") == []


# A skill is shipped verbatim inside every client's package, at whatever path
# that client installs skills to. A reference from a skill to anything outside
# its own folder — the adapter's hooks, a manifest, a relative path upward —
# points at a file that exists in one package layout and nowhere else
# (plugin/PACKAGING.md).
OUTSIDE_THE_SKILL = re.compile(r"\.\./|plugin/|hooks/|commands/|\.claude-plugin|\bplugin\.json\b|\bhooks\.json\b")


def test_the_skills_reference_nothing_outside_their_folder():
    offenders = {
        str(p.relative_to(ROOT)): sorted(set(OUTSIDE_THE_SKILL.findall(p.read_text())))
        for p in sorted((ROOT / "plugin/skills").glob("*/SKILL.md"))
    }
    offenders = {path: refs for path, refs in offenders.items() if refs}
    assert not offenders, (
        f"skills that reference files outside their own folder: {offenders}. A "
        f"skill ships verbatim in every client package; see plugin/PACKAGING.md."
    )


def test_the_layout_guard_can_fail():
    assert OUTSIDE_THE_SKILL.findall("run ../hooks/sync.sh") == ["../", "hooks/"]
    assert OUTSIDE_THE_SKILL.findall("see plugin.json") == ["plugin.json"]
    assert OUTSIDE_THE_SKILL.findall("record the plugin's behaviour") == []