multi-agent-mux/tests/test_issue42_skill_blocks.py

"""Issue #42 stages 1-2: the SKILL.md code-block inventory is data, and the docs follow it.

tests/data/skill_code_blocks.json lists every fenced block of the eight
SKILL.md files with its class - (b) a procedure no script offers yet, (c) an
example, call or diagram - and the (a) blocks removed because the entry script
already does them. These tests fail when a SKILL.md block changes without the
inventory, when a deleted block's body comes back (by exact sha or by a
whitespace-insensitive fingerprint, whatever the inventory says about it), or
when an exit-code table drifts from the entry script. A (c) block folded into
another call is recorded under "superseded".
"""
import hashlib
import json
import re
from pathlib import Path

import pytest

REPO = Path(__file__).resolve().parent.parent
SKILLS = REPO / ".agents" / "skills"
DATA = json.loads((Path(__file__).parent / "data" / "skill_code_blocks.json").read_text(encoding="utf-8"))
SKILL_NAMES = sorted(p.parent.name for p in SKILLS.glob("multi-agent-mux-*/SKILL.md"))
KINDS = {"invocation", "example", "diagram", "procedure", "verification"}

_OPEN = re.compile(r"^(\s*)(`{3,}|~{3,})\s*([^`\s]*)")


def parse_blocks(text):
    """Fenced blocks as (start, end, lang, sha, body); 1-based fence lines.

    Indented fences (inside list items) count: they render as code blocks too.
    A fence closes on a line of the same character, at least as long."""
    lines = text.split("\n")
    out, i = [], 0
    while i < len(lines):
        m = _OPEN.match(lines[i])
        if not m:
            i += 1
            continue
        fence = m.group(2)
        close = re.compile(r"^\s*" + re.escape(fence[0]) + "{%d,}\\s*$" % len(fence))
        j = i + 1
        while j < len(lines) and not close.match(lines[j]):
            j += 1
        body = "\n".join(lines[i + 1:j])
        out.append((i + 1, j + 1, m.group(3) or "-", hashlib.sha256(body.encode()).hexdigest()[:12], body))
        i = j + 1
    return out


def skill_text(skill):
    return (SKILLS / skill / "SKILL.md").read_text(encoding="utf-8")


_HEREDOC = re.compile(r"(?<!<)<<-?\s*(['\"]?)(\w+)\1")
_EXIT = re.compile(r"(?<![\w.])exit\s+(\d+)\b")


_RAISE = re.compile(r"SystemExit\((\d+)\)")


def _scan_exits(script):
    """(`exit N` codes in shell context, {code: line} of `SystemExit(N)` in heredoc bodies).

    Only the entry file itself (not lib.sh or sub-scripts). Comment lines are
    skipped, and `sys.exit(N)` never matches `exit N`. A heredoc's SystemExit is
    the exit code of the unguarded Python step under set -e."""
    codes, raised, pending = set(), {}, []
    for n, line in enumerate((SKILLS / script).read_text(encoding="utf-8").split("\n"), 1):
        if pending:
            if line.strip() == pending[0]:
                pending.pop(0)
            else:
                for code in _RAISE.findall(line):
                    raised.setdefault(code, n)
            continue
        if re.match(r"^\s*#", line):
            continue
        codes.update(_EXIT.findall(line))
        pending.extend(m.group(2) for m in _HEREDOC.finditer(line))
    return codes, raised


def literal_exit_codes(script):
    """Codes the entry script exits with literally: `exit N` in shell context."""
    return _scan_exits(script)[0]


def exit_table(skill):
    """First-column values of the SKILL.md `| Exit | Meaning |` table, in order."""
    lines = skill_text(skill).split("\n")
    heads = [i for i, l in enumerate(lines) if re.match(r"^\|\s*Exit\s*\|\s*Meaning\s*\|", l)]
    assert len(heads) == 1, f"{skill}: expected one '| Exit | Meaning |' table, found {len(heads)}"
    rows = []
    for l in lines[heads[0] + 2:]:
        if not l.startswith("|"):
            break
        rows.append(l.split("|")[1].strip())
    return rows


def _evidence_line(ref):
    path, rng = ref.rsplit(":", 1)
    first, _, last = rng.partition("-")
    return SKILLS / path, int(first), int(last or first)


# 1. The inventory and the docs are the same.
@pytest.mark.parametrize("skill", SKILL_NAMES)
def test_inventory_matches_skill_md(skill):
    actual = [b[:4] for b in parse_blocks(skill_text(skill))]
    listed = [(b["start"], b["end"], b["lang"], b["sha"]) for b in DATA["blocks"] if b["skill"] == skill]
    assert listed == actual, (
        f"{skill}/SKILL.md blocks differ from tests/data/skill_code_blocks.json.\n"
        f"actual (start, end, lang, sha): {json.dumps(actual)}")


def test_inventory_covers_eight_skills():
    assert len(SKILL_NAMES) == 8
    assert {b["skill"] for b in DATA["blocks"]} == set(SKILL_NAMES)


# 2. No (a) block is left.
def test_no_duplicate_block_left():
    for b in DATA["blocks"]:
        assert b["class"] in ("b", "c"), b
        assert b["kind"] in KINDS, b
    for r in DATA["removed"]:
        assert r["class"] == "a", r


def fingerprint(body):
    """A deleted body's identity that survives reindenting and reflowing."""
    return hashlib.sha256(" ".join(body.split()).encode()).hexdigest()[:12]


def _returned_bodies(texts, gone):
    """(skill, start, which) for each current block whose body is a deleted one."""
    hits = []
    for skill, text in texts.items():
        for start, _end, _lang, sha, body in parse_blocks(text):
            for g in gone:
                if sha == g["base_sha"] or fingerprint(body) == g["base_fingerprint"]:
                    hits.append((skill, start, f"{g['skill']}:{g['base_start']}"))
    return hits


# 2b. A deleted body stays deleted - even when the inventory is updated to list
# it again under another class (review R3).
def test_no_deleted_body_came_back():
    gone = DATA["removed"] + DATA["superseded"]
    texts = {s: skill_text(s) for s in SKILL_NAMES}
    assert _returned_bodies(texts, gone) == []


def test_a_returned_body_is_caught_reindented_and_with_the_inventory_updated():
    """The guard itself: a current block posing as a deleted one is found by its
    exact sha and, reindented, by its fingerprint; listing it in the inventory
    does not hide it (that check never reads DATA["blocks"])."""
    skill = "multi-agent-mux-stop"
    start, _end, _lang, sha, body = parse_blocks(skill_text(skill))[0]
    fake = {"skill": skill, "base_start": 1, "base_sha": sha, "base_fingerprint": fingerprint(body)}
    assert _returned_bodies({skill: skill_text(skill)}, [fake]) == [(skill, start, f"{skill}:1")]
    reindented = "\n".join("    " + line for line in body.split("\n"))
    other = {"skill": skill, "base_start": 1, "base_sha": "0" * 12, "base_fingerprint": fingerprint(reindented)}
    assert _returned_bodies({skill: skill_text(skill)}, [other]) == [(skill, start, f"{skill}:1")]


# 2c. A deleted (c) block folded into another call is recorded too (review C1):
# with it the 37 base blocks are all accounted for.
def test_superseded_blocks_are_recorded_and_replaced():
    assert DATA["superseded"], "the folded create call is recorded"
    for r in DATA["superseded"]:
        assert r["class"] == "c" and r["kind"] in KINDS, r
        bodies = [b[4] for b in parse_blocks(skill_text(r["skill"]))]
        assert any(r["replacement"] in body for body in bodies), r
    for r in DATA["removed"] + DATA["superseded"]:
        assert re.fullmatch(r"[0-9a-f]{12}", r["base_fingerprint"]), r


# 3. A removed block is replaced by the entry-point call.
@pytest.mark.parametrize("removed", DATA["removed"], ids=lambda r: f"{r['skill']}:{r['base_start']}")
def test_removed_block_replaced_by_entry_call(removed):
    assert (SKILLS / removed["script"]).is_file(), removed["script"]
    assert removed["replacement"].startswith("bash .agents/skills/" + removed["script"])
    bodies = [b[4] for b in parse_blocks(skill_text(removed["skill"]))]
    assert any(removed["replacement"] in body for body in bodies), \
        f"{removed['skill']}: no block contains {removed['replacement']!r}"
    assert removed["script"] in DATA["entry_points"]


# 4. Exit-code tables do not contradict the code.
@pytest.mark.parametrize("script", sorted(DATA["entry_points"]))
def test_exit_table_agrees_with_script(script):
    ep = DATA["entry_points"][script]
    listed = ep["exit_codes"]
    assert exit_table(ep["skill"]) == listed, f"{ep['skill']}/SKILL.md exit table rows != data"
    assert listed[-1] == "other", "the table needs a catch-all row: under set -e a failing step's code passes through"
    numeric = set(listed[:-1])
    literal = literal_exit_codes(script)
    assert literal <= numeric, f"{script} exits with {sorted(literal - numeric)} but the table lacks them"
    passthrough = ep["passthrough"]
    raised = _scan_exits(script)[1]
    assert {c: f"{script}:{n}" for c, n in raised.items()} == passthrough, \
        f"{script}: SystemExit codes in heredocs {raised} must be declared as passthrough (and in the table)"
    assert set(passthrough) <= numeric, \
        f"{script}: passthrough codes {sorted(set(passthrough) - numeric)} missing from the exit table"
    assert numeric <= literal | set(passthrough), \
        f"table codes {sorted(numeric - literal - set(passthrough))} come from nowhere in {script}"
    for code, ref in passthrough.items():
        path, first, _ = _evidence_line(ref)
        line = path.read_text(encoding="utf-8").split("\n")[first - 1]
        assert re.search(rf"SystemExit\({code}\)|\bexit\s+{code}\b", line), f"{ref} does not raise {code}: {line!r}"


def test_literal_exit_rule_on_base_scripts():
    """The counting rule: heredocs and comments skipped, so create's header `exit 4`
    and stop's `sys.exit(7)` are not counted."""
    assert literal_exit_codes("multi-agent-mux-resume/scripts/resume_session.sh") == {"0", "1", "2"}
    assert literal_exit_codes("multi-agent-mux-stop/scripts/stop_session.sh") == {"0", "1", "2", "3", "4"}
    assert literal_exit_codes("multi-agent-mux-create/scripts/create_session.sh") == {"0", "1", "2", "3", "42", "43"}


# 5. Evidence exists.
def test_evidence_points_into_existing_files():
    for b in DATA["blocks"] + DATA["removed"]:
        for ref in b["evidence"]:
            path, first, last = _evidence_line(ref)
            assert path.is_file(), ref
            assert 1 <= first <= last <= len(path.read_text(encoding="utf-8").split("\n")), ref


# 6. The loading path is unchanged.
@pytest.mark.parametrize("skill", SKILL_NAMES)
def test_frontmatter_name_matches_directory(skill):
    m = re.search(r"^name:\s*(\S+)\s*$", skill_text(skill).split("---")[1], re.M)
    assert m and m.group(1) == skill


def test_claude_skills_link_resolves_to_agents_skills():
    link = REPO / ".claude" / "skills"
    if not link.exists():
        pytest.skip(".claude/skills is untracked; absent in a worktree")
    assert link.resolve() == SKILLS.resolve()