assets/sharded-note.md
> **This document is sharded.** The shard files linked below are the document: they are the source of truth and are what gets maintained. Read this map, then open only the shards you need. `.source/` holds the original as it was before sharding; it is frozen and out of date, so do not load it. To change the document, use the `bmad-shard-pro` skill, which keeps this map, the shards and their checks in step.
bmod.toml
[skill]
bmod = "bmod-core-tools"
source = "github:bmad-code-org/BMAD-METHOD/skills"
recommended_skills = ["bmad-ingest"]
evals/cases.json
{
"cases": [
{
"id": "kestrel-manual",
"input": "Run headless. Our agents need to work from this compressor service manual but it's too long and the headings are just numbers: kestrel-k200-manual.md. Sort it out so they can find things. I won't be around, so don't wait for me.",
"files": [
"https://github.com/bmad-code-org/bmad-method-sample-data/blob/59de7a03913129cff4c6808f2445f4da00ea9b31/shardtest/kestrel-k200-manual/kestrel-k200-manual.md"
],
"rubric": [
"Every line of the original manual is still present word for word in the workspace output (either the original file kept intact somewhere, or files whose contents together contain every original line), checked by sampling the text of paragraphs 2-6a, 3-4a, 5-5 and Table C-3",
"An entry file exists whose first 200 lines let a reader locate the oil level check procedure (paragraph 3-4a, the one saying to wait 10 minutes) and Table C-3 wear limits without reading anything else first: either a contents list with line ranges, or a map linking to the files that hold them",
"If the output is split into several files, no file is named by a bare identifier alone (such as 3-4.md, 5.md or c-2.md); each name says what it covers",
"The output does not point readers to 5-6 for oil separator element replacement or to Table C-2 for rotor and coupling wear limits (the manual's index is wrong on both; the body has 5-5 and Table C-3)",
"The final reply to the user reports at least one of the two index errors (separator element listed at 5-6 but found at 5-5; wear limits listed at Table C-2 but found in Table C-3)",
"No statement in the output files gives a value, limit or part number that differs from the manual (check the 10 minute wait, 3 liters between marks, 19.0 mm coupling minimum, part 60-1140)"
]
},
{
"id": "meridian-proposal",
"input": "Run headless. Agents keep reading our client proposal, meridian-proposal.md. Should it be sharded for them? Do whatever is right and tell me what you did; I'm not around to answer questions.",
"files": [
"https://github.com/bmad-code-org/bmad-method-sample-data/blob/59de7a03913129cff4c6808f2445f4da00ea9b31/shardtest/meridian-proposal/meridian-proposal.md"
],
"rubric": [
"The final reply recommends against splitting the proposal into multiple files, giving its short length (about 3,000 words) or its whole-document purpose as the reason",
"No shard files were created: the workspace holds no new folder of split section files derived from the proposal",
"meridian-proposal.md either is byte-identical to the fixture or differs only by an added contents list near the top; no wording of the proposal itself was changed"
]
},
{
"id": "tidewater-prd",
"input": "Run headless. Our agents have to work from this huge PRD, tidewater-prd.md (about 134k words). Make it navigable for them. I'm away, so decide yourself and tell me what you did.",
"files": [
"https://github.com/bmad-code-org/bmad-method-sample-data/blob/59de7a03913129cff4c6808f2445f4da00ea9b31/shardtest/tidewater-prd/tidewater-prd.md"
],
"rubric": [
"Every line of the original PRD is still present word for word in the workspace output (the original kept intact somewhere, or files whose contents together contain every original line), checked by sampling FR-WO-015, NFR-RET-004, Appendix A and the section 5.3 plan table",
"An entry file exists whose first 200 lines let a reader locate where FR-WO-015 and the Appendix A traceability matrix are, without reading anything else first",
"The final reply or a conflicts list in the output reports that the Professional price is stated as both $27 and $29, and that raw telematics retention is stated as both 18 and 25 months, without asserting only one value",
"No output file other than a copy of the original rewrites PRD content in new words: requirement text in the output matches the source verbatim"
]
}
]
}
evals/triggers.json
[
{
"query": "this PRD is like 400 pages and our agents keep choking on it, can you shard it? it's prd-fleet/prd-fleet.md",
"should_trigger": true
},
{
"query": "I've got a six hour meeting transcript (offsite.vtt) — turn it into something my coding agents can actually look things up in",
"should_trigger": false
},
{
"query": "our employee handbook is a 90 page handbook.docx and the support bot can't read it all, make it navigable for the bot",
"should_trigger": true
},
{
"query": "add a table of contents with line ranges to architecture.md so agents can jump straight to the section they need",
"should_trigger": true
},
{
"query": "I exported our #product-mobile slack channel (folder slack-export/), can you turn it into a wiki our agents can use",
"should_trigger": false
},
{
"query": "split this compressor service manual into smaller files with a map — the headings are just numbers like 3-4 and C-2 so a plain split is useless",
"should_trigger": true
},
{
"query": "update the sharded PRD in prd-fleet/: add a requirement that admins can force SSO for all users",
"should_trigger": true
},
{
"query": "the architecture doc in architecture-acme/ has grown to 90k words. is it worth breaking up for agents or should I leave it?",
"should_trigger": true
},
{
"query": "here's a folder of 15 weekly project notes from three different note takers, organise them into one wiki agents can navigate",
"should_trigger": false
},
{
"query": "specs/riverton-rfp.pdf is way too long for agents to work from, do something about it",
"should_trigger": true
},
{
"query": "add a table of contents to the README so it renders nicely on GitHub",
"should_trigger": false
},
{
"query": "summarise this 40 page PRD into one page for the exec team meeting tomorrow",
"should_trigger": false
},
{
"query": "break this PRD down into epics and stories for the sprint",
"should_trigger": false
},
{
"query": "review the architecture document for gaps and risks before we start building",
"should_trigger": false
},
{
"query": "create a PRD for our fleet maintenance app, I have notes from the customer calls",
"should_trigger": false
},
{
"query": "process this research report into something the team can decide from",
"should_trigger": false
},
{
"query": "this python file is 3000 lines, split it into smaller modules",
"should_trigger": false
},
{
"query": "split users.csv into files of 1000 rows each",
"should_trigger": false
},
{
"query": "translate this meeting transcript into Spanish for the Madrid team",
"should_trigger": false
},
{
"query": "fix the broken links in our docs wiki after the folder rename",
"should_trigger": false
}
]
references/update.md
# Updating a prepared document
The user changes the document through its main file. The outcome is the document changed as they asked, still passing `check`, with every change in the memlog. The shards are the document now, and `.source/` stays frozen, so never edit or read from it.
The script is `uv run {skill-root}/scripts/shard_tool.py <command>`; the memlog script is `uv run {project-root}/_bmad/scripts/memlog.py`. `{doc_folder}` is the folder holding the main file, or the file with the contents list.
Before applying a change, find where the document already speaks to it, using the map, the identifier lookup and search. When the change contradicts what is there, show the user both and let them decide. When it touches several shards, change all of them.
## A sharded folder
- **New content with a home:** add it to the shard where a reader would look first.
- **New content with no home, or a shard grown past about 14 KB:** split it or add a new shard. Name the new shard from its content, never starting with the document type's prefix.
- **Record each new, renamed or re-summarised shard in `.shards.json`:** file, title and a one-line summary, placed in its map group. Then run `map` to rewrite the main file and its sub-maps.
- **Identifiers:** when the document has an identifier lookup, run `lookup` after any change that adds, moves or removes one.
Run `check` without `--rebuild`, since the shards have moved on from the original and no longer join back into it, and fix every failure.
## A file with a contents list
- **New section:** run `toc <file> --add <line> --title <title>`, where `<line>` is the section's first line as the file reads now; add `--group` when it belongs to a different group than the section before it.
- **Otherwise:** run `toc <file>` to refresh the line ranges. When it reports a lost section (its first line was renamed or removed), re-add it with `--add`, or leave it out if the section is gone. When it reports an ambiguous one (its first line now appears in more places), plan the list again with `--plan`.
Then run `check <file>` and fix every failure.
When the document has outgrown a contents list, tell the user and offer a faithful split; remove the list with `toc <file> --remove` before planning one.
## Record
Append each change to `{doc_folder}/.memlog.md` with `memlog.py append --type change`, and each choice the user made with `--type decision`; run `memlog.py init --workspace {doc_folder}` first when none exists. Without `_bmad/`, skip memlog writes and say so once.
scripts/shard_tool.py
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# ///
"""shard_tool — the deterministic half of bmad-shard-pro.
Every subcommand prints one JSON object to stdout; diagnostics go to stderr.
Exit codes: 0 = done or clean, 1 = check failures, 2 = usage or input error.
Subcommands:
outline SOURCE [--depth N] [--from L --to L] [--ids]
Size, headings, bare-identifier headings, tables and fenced blocks, frontmatter,
an existing contents list, identifier prefixes, and a proposed cut along headings
sized by characters that never splits a table or fenced block.
build PLAN
Build the sharded folder from a plan: cut faithful shards word for word, write the main
file (note + map, sub-maps when the map would pass line 200), move the source into
`.source/`, save the plan as `.shards.json`, and write the identifier lookup.
map FOLDER
Rewrite the main file's map (and sub-maps) from `.shards.json` after shards were added,
renamed or re-summarised; the main file's frontmatter is kept.
lookup FOLDER
Regenerate `identifier-lookup.md` from the shards, using the plan's lookup pattern.
toc FILE [--plan PLAN]
Insert a contents list after the title, or with no plan refresh the line ranges of the
one already there.
check TARGET [--rebuild]
Check a sharded folder or a file with a contents list. --rebuild also checks that the
faithful shards join back into the source exactly (only true before any update).
Input is markdown or plain text; anything else is converted before this tool sees it.
Plan JSON: see `build --help`.
"""
from __future__ import annotations
import argparse
import json
import re
import shutil
import sys
from collections import Counter
from pathlib import Path
PLAN_HELP = """Plan JSON:
{
"source": "the markdown or text file being split",
"folder": "the document folder; the main file is <folder>/<folder name>.md",
"title": "title for the main file",
"lookup": "regex matching identifiers to index (optional)",
"groups": [
{"heading": "Map group", "summary": "optional, used when the map is split",
"shards": [
{"file": "name.md", "title": "Title", "summary": "One line", "start": 1, "end": 40}
]}
]
}
Shards are cut word for word from start to end (1-based, inclusive); together they cover
every line once, in order. In a toc plan only start and title are read.
"""
NOTE_PATH = Path(__file__).resolve().parent.parent / "assets" / "sharded-note.md"
NOTE_MARK = "This document is sharded."
MANIFEST = ".shards.json"
LOOKUP_FILE = "identifier-lookup.md"
WINDOW = 200
MIN_CHARS = 3000
MAX_CHARS = 14000
TOC_START = "<!-- shard-pro:contents -->"
TOC_END = "<!-- /shard-pro:contents -->"
HEADING = re.compile(r"^(#{1,6})\s+(.*?)\s*#*\s*$")
FENCE = re.compile(r"^\s{0,3}(`{3,}|~{3,})(.*)$")
NUMBERED = re.compile(r"^\s{0,3}\d+[.)]\s")
BARE_ID = re.compile(r"^[A-Z]?\d+(?:[-.]\d+)*[a-z]?$|^[A-Z](?:-\d+[a-z]?)*$")
IDENT = re.compile(r"\b[A-Z][A-Z0-9]{0,5}(?:-[A-Z0-9]{1,6})*-\d{1,4}[a-z]?\b")
LINK = re.compile(r"\]\(\./([^)#\s]+)")
ANY_LINK = re.compile(r"\]\(([^)\s]+)\)")
def emit(obj: dict, code: int = 0) -> int:
print(json.dumps(obj, indent=1, ensure_ascii=False))
return code
def fail(msg: str) -> int:
print(msg, file=sys.stderr)
return emit({"ok": False, "error": msg}, 2)
def read_lines(path: Path) -> list[str]:
return path.read_text(encoding="utf-8").splitlines()
def frontmatter_end(lines: list[str]) -> int:
"""Number of lines taken by YAML frontmatter at the top (0 when none)."""
if lines and lines[0].strip() == "---":
for i in range(1, len(lines)):
if lines[i].strip() == "---":
return i + 1
return 0
def block_spans(lines: list[str]) -> list[tuple[int, int, str]]:
"""1-based inclusive spans of fenced blocks (kind = fence language or 'code'), tables and numbered steps."""
spans, i, n = [], 0, len(lines)
while i < n:
m = FENCE.match(lines[i])
if m:
lang = m.group(2).strip() or "code"
close = re.compile(rf"^\s{{0,3}}{re.escape(m.group(1)[0])}{{{len(m.group(1))},}}\s*$")
j = i + 1
while j < n and not close.match(lines[j]):
j += 1
spans.append((i + 1, min(j, n - 1) + 1, lang))
i = j + 1
continue
if NUMBERED.match(lines[i]):
j = i
while j + 1 < n:
nxt = lines[j + 1]
if NUMBERED.match(nxt) or (nxt.startswith((" ", "\t")) and nxt.strip()):
j += 1
elif (
not nxt.strip()
and j + 2 < n
and (NUMBERED.match(lines[j + 2]) or lines[j + 2].startswith((" ", "\t")))
):
j += 2 # a loose list: one blank line between items
else:
break
if j > i:
spans.append((i + 1, j + 1, "steps"))
i = j + 1
continue
if lines[i].lstrip().startswith("|"):
j = i
while j + 1 < n and lines[j + 1].lstrip().startswith("|"):
j += 1
if j > i:
spans.append((i + 1, j + 1, "table"))
i = j + 1
continue
i += 1
return spans
def headings(lines: list[str], spans: list[tuple[int, int, str]]) -> list[dict]:
inside = set()
for a, b, kind in spans:
if kind not in ("table", "steps"):
inside.update(range(a, b + 1))
out = []
for i, line in enumerate(lines, 1):
m = HEADING.match(line)
if m and i not in inside:
text = m.group(2)
h = {"line": i, "level": len(m.group(1)), "text": text, "bare": bool(BARE_ID.match(text))}
if h["bare"]:
# A bare identifier says nothing; the first line of text under it names the subject.
first = next((x.strip() for x in lines[i:] if x.strip() and not HEADING.match(x)), "")
h["first_line"] = first[:100]
out.append(h)
return out
def chars(lines: list[str], a: int, b: int) -> int:
return sum(len(x) + 1 for x in lines[a - 1 : b])
def propose_cut(lines: list[str], heads: list[dict], spans: list[tuple[int, int, str]], start_line: int) -> list[dict]:
"""Cut at headings: split oversized sections at deeper headings, merge small neighbours."""
n = len(lines)
blocked = set()
for a, b, _ in spans:
blocked.update(range(a + 1, b + 1))
cuttable = [h for h in heads if h["line"] not in blocked and h["line"] > start_line]
def split(a: int, b: int, level: int) -> list[tuple[int, int]]:
if chars(lines, a, b) <= MAX_CHARS or level > 6:
return [(a, b)]
points = [h["line"] for h in cuttable if a < h["line"] <= b and h["level"] == level]
if not points:
return split(a, b, level + 1)
bounds = [a] + points + [b + 1]
out = []
for x, y in zip(bounds, bounds[1:], strict=False):
if y - 1 >= x:
out.extend(split(x, y - 1, level + 1))
return out
def by_size(a: int, b: int) -> list[tuple[int, int]]:
"""Cut a span with no headings at line boundaries, outside tables and fenced blocks."""
out, start, size = [], a, 0
for i in range(a, b + 1):
size += len(lines[i - 1]) + 1
if size >= MAX_CHARS * 0.7 and i < b and i + 1 not in blocked:
out.append((start, i))
start, size = i + 1, 0
return out + [(start, b)]
top = min((h["level"] for h in cuttable), default=1)
segs = split(start_line, n, top) if n >= start_line else []
segs = [piece for a, b in segs for piece in (by_size(a, b) if chars(lines, a, b) > MAX_CHARS else [(a, b)])]
merged: list[list[int]] = []
for a, b in segs:
if (
merged
and (chars(lines, *merged[-1]) < MIN_CHARS or chars(lines, a, b) < MIN_CHARS)
and chars(lines, merged[-1][0], b) <= MAX_CHARS
):
merged[-1][1] = b
else:
merged.append([a, b])
by_line = {h["line"]: h for h in heads}
out = []
for a, b in merged:
first = next((by_line[i] for i in range(a, b + 1) if i in by_line), None)
out.append(
{
"start": a,
"end": b,
"chars": chars(lines, a, b),
"heading": first["text"] if first else lines[a - 1][:80],
}
)
return out
def find_toc(lines: list[str]) -> dict | None:
for i, line in enumerate(lines[:WINDOW], 1):
if line.strip() == TOC_START:
return {"line": i, "made_by": "shard-pro"}
m = HEADING.match(line)
if m and re.fullmatch(r"(table of )?contents", m.group(2).strip(), re.I):
return {"line": i, "made_by": "source"}
return None
def cmd_outline(args) -> int:
path = Path(args.source).resolve()
if not path.is_file():
return fail(f"not a file: {path}")
lines = read_lines(path)
spans = block_spans(lines)
own = toc_lines(lines)
if own:
spans.append((own[0] + 1, own[1], "contents"))
heads = headings(lines, spans)
fm = frontmatter_end(lines)
prefixes: dict[str, int] = {}
defined: dict[str, int] = {}
for h in heads:
for ident in IDENT.findall(h["text"]):
prefixes[ident.rsplit("-", 1)[0]] = prefixes.get(ident.rsplit("-", 1)[0], 0) + 1
defined.setdefault(ident, []).append(h["line"])
sections = [h for h in heads if h["level"] > 1]
shown = [h for h in heads if h["level"] <= args.depth]
if args.from_line or args.to_line:
lo, hi = args.from_line or 1, args.to_line or len(lines)
shown = [h for h in heads if lo <= h["line"] <= hi]
out = {
"source": str(path),
"lines": len(lines),
"chars": sum(len(x) + 1 for x in lines),
"words": sum(len(x.split()) for x in lines),
"frontmatter_lines": fm,
"heading_count": len(heads),
"bare_heading_share": round(sum(h["bare"] for h in sections) / len(sections), 2) if sections else 0,
"existing_contents": find_toc(lines),
"blocks": [{"start": a, "end": b, "kind": k} for a, b, k in spans if k != "table" or b - a >= 3],
"identifier_prefixes": dict(sorted(prefixes.items(), key=lambda kv: -kv[1])[:20]),
"headings": [
{k: h[k] for k in ("line", "level", "text")} | ({"first_line": h["first_line"]} if h["bare"] else {})
for h in shown
],
"proposed_cut": propose_cut(lines, heads, spans, fm + 1),
}
twice = {i: ls for i, ls in defined.items() if len(ls) > 1}
if twice:
out["identifiers_defined_twice"] = twice
if args.ids:
out["defined_identifiers"] = defined
return emit(out)
def all_shards(plan: dict) -> list[dict]:
return [s for g in plan["groups"] for s in g["shards"]]
def slug(text: str) -> str:
return re.sub(r"[^a-z0-9]+", "-", text.lower()).strip("-") or "group"
def map_entry(s: dict) -> str:
return f"- [{s['title']}](./{s['file']}) — {s['summary']}"
def main_text(plan: dict, front: list[str], folder: Path, has_lookup: bool) -> tuple[str, dict[str, str]]:
"""The main file, plus sub-map files when the full map would pass the 200-line window."""
note = NOTE_PATH.read_text(encoding="utf-8").strip()
head = front + [f"# {plan['title']}", "", note, "", "## Map", ""]
lookup = [f"- [Identifier lookup](./{LOOKUP_FILE}) — which shard holds each identifier"] if has_lookup else []
body = lookup + ([""] if lookup else [])
for g in plan["groups"]:
body += [f"### {g['heading']}", ""] + [map_entry(s) for s in g["shards"]] + [""]
if len(head) + len(body) <= WINDOW:
return "\n".join(head + body).rstrip() + "\n", {}
subs, short = {}, list(lookup)
for g in plan["groups"]:
name, k = f"map-{slug(g['heading'])}.md", 2
while name in subs:
name, k = f"map-{slug(g['heading'])}-{k}.md", k + 1
about = g.get("summary") or "; ".join(s["title"] for s in g["shards"][:3])
short.append(f"- [{g['heading']}](./{name}) — {len(g['shards'])} shards: {about}")
lines = [f"# {plan['title']}: {g['heading']}", "", f"Part of [{plan['title']}](./{folder.name}.md).", ""]
subs[name] = "\n".join(lines + [map_entry(s) for s in g["shards"]]) + "\n"
return "\n".join(head + short).rstrip() + "\n", subs
def validate_plan(plan: dict, lines: list[str]) -> str | None:
n_lines = len(lines)
for key in ("source", "folder", "title", "groups"):
if key not in plan:
return f"plan is missing '{key}'"
if plan.get("lookup"):
try:
re.compile(plan["lookup"])
except re.error as e:
return f"lookup pattern does not compile: {e}"
inside = {}
for a, b, kind in block_spans(lines):
for i in range(a + 1, b + 1):
inside[i] = (a, b, kind)
for g in plan["groups"]:
if "heading" not in g or not isinstance(g.get("shards"), list):
return f"each group needs a heading and a shards list: {str(g)[:80]}"
if len(g["shards"]) > WINDOW - 4:
return f"group '{g['heading']}' has {len(g['shards'])} shards; a map page holds {WINDOW - 4}, so split the group"
if len(plan["groups"]) > WINDOW - 10:
return f"{len(plan['groups'])} groups will not fit one map page; merge some"
if not all_shards(plan):
return "the plan has no shards"
fm = frontmatter_end(lines)
seen, expect = set(), None
for s in all_shards(plan):
for key in ("file", "title", "summary", "start", "end"):
if key not in s:
return f"shard {s.get('file', '?')} is missing '{key}'"
if s["file"] in seen:
return f"duplicate shard file {s['file']}"
if Path(s["file"]).name != s["file"] or not s["file"].endswith(".md"):
return f"shard file {s['file']} must be a plain .md file name, with no folder"
seen.add(s["file"])
if not isinstance(s["start"], int) or not isinstance(s["end"], int):
return f"{s['file']} needs whole-number start and end lines"
if expect is None:
if s["start"] not in (1, fm + 1):
return f"{s['file']} starts at line {s['start']}; the first shard starts at line {fm + 1}"
expect = s["start"]
if s["start"] != expect:
return f"{s['file']} starts at line {s['start']}; expected {expect} (gap or overlap)"
if s["end"] < s["start"]:
return f"{s['file']} ends before it starts"
if s["start"] in inside:
a, b, kind = inside[s["start"]]
return f"{s['file']} starts at line {s['start']}, inside the {kind} block at lines {a}-{b}"
expect = s["end"] + 1
if expect is not None and expect != n_lines + 1:
return f"faithful ranges end at line {expect - 1}; the source has {n_lines}"
return None
def clashes(plan: dict, folder: Path, src: Path) -> str | None:
"""Why build would overwrite a file it did not write, or write two outputs to one path."""
main = f"{folder.name}.md"
_, subs = main_text(plan, [], folder, bool(plan.get("lookup")))
reserved = {main, MANIFEST} | set(subs) | ({LOOKUP_FILE} if plan.get("lookup") else set())
for s in all_shards(plan):
if s["file"] in reserved:
return f"shard file {s['file']} would collide with a file shard-pro writes; rename it"
old = load_manifest(folder) or {"groups": []}
ours = {x["file"] for x in all_shards(old)} | set(old.get("sub_maps", [])) | {LOOKUP_FILE, MANIFEST}
for name in sorted({s["file"] for s in all_shards(plan)} | reserved):
p = folder / name
if not p.exists() or name in ours:
continue
if name == main and (p.resolve() == src.resolve() or any(NOTE_MARK in x for x in read_lines(p)[:WINDOW])):
continue
return (
f"{name} already exists in {folder.name}/ and shard-pro did not write it; rename the shard or move the file"
)
return None
def cmd_build(args) -> int:
plan_path = Path(args.plan).resolve()
plan = json.loads(plan_path.read_text(encoding="utf-8"))
src = Path(plan.get("source", ""))
folder = Path(plan.get("folder", "")).resolve()
if (folder / ".source" / src.name).is_file():
# A rebuild: the frozen copy is the source, even where the main file now sits at the old path.
src = folder / ".source" / src.name
if not src.is_file():
return fail(f"source not found: {src}")
lines = read_lines(src)
if toc_lines(lines):
return fail(
f"{src.name} carries a shard-pro contents list; remove it with `toc {src} --remove`, then outline and plan again"
)
err = validate_plan(plan, lines) or clashes(plan, folder, src)
if err:
return fail(err)
folder.mkdir(parents=True, exist_ok=True)
old = load_manifest(folder)
if old:
keep = {x["file"] for x in all_shards(plan)}
for name in [x["file"] for x in all_shards(old)] + old.get("sub_maps", []) + [LOOKUP_FILE]:
if name not in keep:
(folder / name).unlink(missing_ok=True)
store = folder / ".source"
store.mkdir(exist_ok=True)
fm = frontmatter_end(lines)
front = lines[:fm]
if src.resolve() != (store / src.name).resolve():
shutil.copy2(src, store / src.name)
written = []
for s in all_shards(plan):
a = max(s["start"], fm + 1)
header = [f"# {s['title']}", "", f"Part of [{plan['title']}](./{folder.name}.md).", ""]
body = lines[a - 1 : s["end"]]
(folder / s["file"]).write_text("\n".join(header + body) + "\n", encoding="utf-8")
s["header_lines"] = len(header)
written.append({"file": s["file"], "chars": sum(len(x) + 1 for x in body)})
has_lookup = bool(plan.get("lookup"))
text, subs = main_text(plan, front, folder, has_lookup)
main = folder / f"{folder.name}.md"
main.write_text(text, encoding="utf-8")
for name, body in subs.items():
(folder / name).write_text(body, encoding="utf-8")
plan["source"] = f".source/{src.name}"
plan["frontmatter_lines"] = fm
plan["sub_maps"] = sorted(subs)
(folder / MANIFEST).write_text(json.dumps(plan, indent=1, ensure_ascii=False) + "\n", encoding="utf-8")
if src.exists() and src.resolve() != main.resolve() and src.resolve().parent != store.resolve():
src.unlink()
lookup = write_lookup(folder, plan) if has_lookup else None
return emit(
{
"ok": True,
"main": str(main),
"shards_cut": written,
"sub_maps": sorted(subs),
"lookup": lookup,
}
)
def body_lines(path: Path) -> list[str]:
"""A shard's lines without its own title, so identifiers in a title are not read as defined there."""
lines = read_lines(path)
return lines[1:] if lines and lines[0].startswith("# ") else lines
def defines(line: str) -> bool:
"""A line that introduces an identifier: a heading or a bold label."""
return bool(HEADING.match(line)) or line.lstrip().startswith("**")
def identifiers(pattern: re.Pattern, line: str) -> list[str]:
return [m.group(0) for m in pattern.finditer(line)]
def write_lookup(folder: Path, plan: dict) -> dict:
pattern = re.compile(plan["lookup"])
homes: dict[str, list[str]] = {}
for s in all_shards(plan):
p = folder / s["file"]
if not p.exists():
continue
for line in body_lines(p):
if defines(line):
for ident in identifiers(pattern, line):
if s["file"] not in homes.setdefault(ident, []):
homes[ident].append(s["file"])
rows = [
f"| {i} | {', '.join(f'[{Path(f).stem}](./{f})' for f in fs)} |"
for i, fs in sorted(homes.items(), key=lambda kv: natural(kv[0]))
]
text = [
"# Identifier lookup",
"",
f"Part of [{plan['title']}](./{folder.name}.md). Which shard holds each identifier.",
"",
"| Identifier | Shard |",
"|---|---|",
] + rows
(folder / LOOKUP_FILE).write_text("\n".join(text) + "\n", encoding="utf-8")
return {"identifiers": len(homes), "in_several_shards": sorted(i for i, fs in homes.items() if len(fs) > 1)}
def natural(s: str) -> list:
return [int(t) if t.isdigit() else t for t in re.split(r"(\d+)", s)]
def cmd_lookup(args) -> int:
folder = Path(args.folder).resolve()
plan = load_manifest(folder)
if plan is None:
return fail(f"no {MANIFEST} in {folder}")
if not plan.get("lookup"):
return fail("the plan has no lookup pattern")
return emit({"ok": True} | write_lookup(folder, plan))
def cmd_map(args) -> int:
folder = Path(args.folder).resolve()
plan = load_manifest(folder)
if plan is None:
return fail(f"no {MANIFEST} in {folder}")
main = folder / f"{folder.name}.md"
current = read_lines(main) if main.exists() else []
front = current[: frontmatter_end(current)]
text, subs = main_text(plan, front, folder, bool(plan.get("lookup")))
main.write_text(text, encoding="utf-8")
for old in set(plan.get("sub_maps", [])) - set(subs):
(folder / old).unlink(missing_ok=True)
for name, body in subs.items():
(folder / name).write_text(body, encoding="utf-8")
plan["sub_maps"] = sorted(subs)
(folder / MANIFEST).write_text(json.dumps(plan, indent=1, ensure_ascii=False) + "\n", encoding="utf-8")
return emit({"ok": True, "main": str(main), "sub_maps": sorted(subs)})
def load_manifest(folder: Path) -> dict | None:
p = folder / MANIFEST
return json.loads(p.read_text(encoding="utf-8")) if p.exists() else None
def toc_block(entries: list[dict], offset: int) -> list[str]:
out = [
TOC_START,
"## Contents",
"",
"Line numbers are lines of this file; read a section by its range. To change the document, use the `bmad-shard-pro` skill, which keeps this list true.",
]
group = None
for e in entries:
if e.get("group") and e["group"] != group:
group = e["group"]
out += ["", f"**{group}**"]
out.append(f"- L{e['start'] + offset}-{e['end'] + offset} {e['title']}")
return out + [TOC_END]
def title_end(body: list[str]) -> int:
"""Index after the frontmatter and the title line: where the contents list goes."""
at = frontmatter_end(body)
return at + 1 if at < len(body) and body[at].startswith("# ") else at
def strip_toc(lines: list[str]) -> list[str]:
"""The file without a shard-pro contents list and the blank line inserted before it."""
marks = [x.strip() for x in lines]
if TOC_START not in marks or TOC_END not in marks:
return lines
a, b = marks.index(TOC_START), marks.index(TOC_END)
if a > 0 and lines[a - 1] == "":
a -= 1
return lines[:a] + lines[b + 1 :]
def toc_lines(raw: list[str]) -> tuple[int, int] | None:
"""0-based start and end (exclusive) of a shard-pro contents list and the blank line before it."""
marks = [x.strip() for x in raw]
if TOC_START not in marks or TOC_END not in marks:
return None
a, b = marks.index(TOC_START), marks.index(TOC_END) + 1
return (a - 1 if a > 0 and raw[a - 1] == "" else a), b
def to_body(raw: list[str], line: int) -> int:
"""A line number of the file as it reads now, counted in the file without its contents list."""
span = toc_lines(raw)
return line - (span[1] - span[0]) if span and line > span[1] else line
def ambiguous_anchors(stored: list[dict], body: list[str]) -> list[dict]:
"""Sections whose first line now appears more often than the list has sections starting with it."""
want = Counter(x["anchor"] for x in stored)
have = Counter(body)
return [
{"first_line": a, "at_lines": [i for i, x in enumerate(body, 1) if x == a]} for a in want if have[a] > want[a]
]
def sidecar(path: Path) -> Path:
return path.with_name(f".{path.stem}.contents.json")
def cmd_toc(args) -> int:
path = Path(args.file).resolve()
raw = read_lines(path)
body = strip_toc(raw)
if args.plan:
plan = json.loads(Path(args.plan).read_text(encoding="utf-8"))
items = [
{"start": s.get("start"), "title": s.get("title"), "group": g.get("heading")}
for g in plan.get("groups", [])
for s in g.get("shards", [])
]
prev = 0
for e in items:
if not e["title"] or not isinstance(e["start"], int):
return fail(f"every entry needs a title and a start line: {e}")
e["start"] = to_body(raw, e["start"])
if not prev < e["start"] <= len(body):
return fail(
f"'{e['title']}' starts at line {e['start']}; starts must rise and stay within the {len(body)}-line file"
)
prev = e["start"]
elif args.remove:
path.write_text("\n".join(body) + "\n", encoding="utf-8")
sidecar(path).unlink(missing_ok=True)
return emit({"ok": True, "file": str(path), "removed": len(raw) - len(body)})
else:
if not sidecar(path).exists():
return fail("no shard-pro contents list for this file; pass --plan to create one")
# Each section is re-found by the text of its first line, which survives edits elsewhere.
stored = json.loads(sidecar(path).read_text(encoding="utf-8"))
twice = ambiguous_anchors(stored, body)
if twice:
return emit({"ok": False, "file": str(path), "ambiguous": twice}, 1)
items, pos, lost = [], 0, []
for item in stored:
idx = next((i for i in range(pos, len(body)) if body[i] == item["anchor"]), None)
if idx is None:
lost.append(item)
continue
items.append({"start": idx + 1, "title": item["title"], "group": item.get("group")})
pos = idx + 1
if args.add:
line = to_body(raw, args.add)
if not args.title or not 1 <= line <= len(body):
return fail("--add needs --title and a line number inside the file")
new_item = {"start": line, "title": args.title, "group": args.group}
if any(e["start"] == line for e in items):
return fail(f"a section already starts at line {args.add}")
items = sorted(items + [new_item], key=lambda e: e["start"])
if new_item["group"] is None:
k = items.index(new_item)
new_item["group"] = items[k - 1]["group"] if k else None
at = title_end(body)
for i, e in enumerate(items):
e["start"] = max(e["start"], at + 1)
limit = items[i + 1]["start"] - 1 if i + 1 < len(items) else len(body)
# A blank line is a poor anchor: start each section at its first line with text.
while e["start"] < limit and not body[e["start"] - 1].strip():
e["start"] += 1
for i, e in enumerate(items):
e["end"] = items[i + 1]["start"] - 1 if i + 1 < len(items) else len(body)
anchors = [body[e["start"] - 1] for e in items]
block = toc_block(items, 0)
block = toc_block(items, len(block) + 1)
new = body[:at] + [""] + block + body[at:]
path.write_text("\n".join(new) + "\n", encoding="utf-8")
side = [{"anchor": a, "title": e["title"], "group": e.get("group")} for a, e in zip(anchors, items, strict=True)]
sidecar(path).write_text(json.dumps(side, indent=1, ensure_ascii=False) + "\n", encoding="utf-8")
end = at + 1 + len(block)
out = {"ok": True, "file": str(path), "entries": len(items), "list_ends_at_line": end, "fits_window": end <= WINDOW}
if not args.plan and lost:
# A renamed or removed section start: the rest are refreshed; the lost ones are re-added with --add.
out |= {"ok": False, "lost": [{"title": x["title"], "first_line_was": x["anchor"]} for x in lost]}
return emit(out, 1)
return emit(out)
def check_toc(path: Path) -> tuple[list, list]:
fails, warns = [], []
lines = read_lines(path)
side = sidecar(path)
if TOC_START not in [x.strip() for x in lines] or not side.exists():
return [{"check": "contents", "file": path.name, "text": "no shard-pro contents list"}], []
stored = json.loads(side.read_text(encoding="utf-8"))
end = next((i for i, x in enumerate(lines, 1) if x.strip() == TOC_END), None)
if end is None:
return [
{
"check": "contents",
"file": path.name,
"text": "the contents list has no end marker; run toc to rebuild it",
}
], []
if end > WINDOW:
fails.append({"check": "window", "file": path.name, "text": f"contents list ends at line {end}, past {WINDOW}"})
ranges = [re.match(r"^- L(\d+)-(\d+) ", x) for x in lines[:end]]
ranges = [(int(m.group(1)), int(m.group(2))) for m in ranges if m]
for (a, b), nxt in zip(ranges, ranges[1:] + [(len(lines) + 1, 0)], strict=True):
if b != nxt[0] - 1:
want = "the end of the file" if nxt[1] == 0 else f"line {nxt[0] - 1}"
fails.append(
{
"check": "contents",
"file": path.name,
"line": a,
"text": f"L{a}-{b} should end at {want}; run toc to refresh",
}
)
prev = 0
for a, b in ranges:
if not prev < a <= b:
fails.append(
{
"check": "contents",
"file": path.name,
"line": a,
"text": f"L{a}-{b} is out of order or ends before it starts",
}
)
prev = b
for item in ambiguous_anchors(stored, strip_toc(lines)):
fails.append(
{
"check": "contents",
"file": path.name,
"text": f"'{item['first_line']}' starts several sections; re-plan the list with toc --plan",
}
)
listed = [a for a, _ in ranges]
for a, item in zip(listed, stored, strict=False):
if a > len(lines) or lines[a - 1] != item["anchor"]:
fails.append(
{
"check": "contents",
"file": path.name,
"line": a,
"text": f"L{a} is not the start of '{item['title']}'; run toc to refresh",
}
)
if len(listed) != len(stored):
fails.append(
{
"check": "contents",
"file": path.name,
"text": "contents list and its record disagree; run toc to refresh",
}
)
return fails, warns
def anchor_slug(text: str) -> str:
return re.sub(r"[^\w\- ]", "", text.strip().lower()).replace(" ", "-")
def broken_links(shard: Path) -> list[dict]:
"""Relative links and #anchors in a shard that no longer reach their target after the split."""
lines = read_lines(shard)
own = {anchor_slug(m.group(2)) for m in map(HEADING.match, lines) if m}
out = []
for i, line in enumerate(lines, 1):
for target in ANY_LINK.findall(line):
if "://" in target or target.startswith(("mailto:", "/", "<")):
continue
path, _, anchor = target.partition("#")
gone = not (shard.parent / path).exists() if path else anchor_slug(anchor) not in own
if gone:
out.append(
{"check": "link", "file": shard.name, "line": i, "text": f"link to {target} no longer resolves"}
)
return out
def check_folder(folder: Path, rebuild: bool) -> tuple[list, list]:
fails, warns = [], []
main = folder / f"{folder.name}.md"
if not main.exists():
return [{"check": "main", "file": main.name, "text": "main file missing"}], []
plan = load_manifest(folder)
if plan is None:
return [
{
"check": "manifest",
"file": MANIFEST,
"text": f"{MANIFEST} missing; this folder was not built by shard-pro",
}
], []
main_lines = read_lines(main)
if not any(NOTE_MARK in x for x in main_lines[:WINDOW]):
fails.append({"check": "note", "file": main.name, "text": "the sharded-document note is missing"})
shards = all_shards(plan)
names = {s["file"] for s in shards}
maps = [main.name] + plan.get("sub_maps", [])
linked_from_maps: set[str] = set()
for m in maps:
p = folder / m
if not p.exists():
fails.append({"check": "map", "file": m, "text": "map file missing"})
continue
ml = read_lines(p)
for i, line in enumerate(ml, 1):
for target in LINK.findall(line):
linked_from_maps.add(target)
if i > WINDOW:
fails.append(
{"check": "window", "file": m, "line": i, "text": f"map link to {target} is past line {WINDOW}"}
)
if plan.get("lookup"):
names.add(LOOKUP_FILE)
for f in sorted(names - linked_from_maps - set(maps)):
fails.append({"check": "orphan", "file": f, "text": "shard is not linked from the map"})
prefix = folder.name.split("-")[0] + "-" if "-" in folder.name else None
for p in sorted(folder.glob("*.md")):
for i, line in enumerate(read_lines(p), 1):
for target in LINK.findall(line):
if not (folder / target).exists():
fails.append({"check": "link", "file": p.name, "line": i, "text": f"link to missing {target}"})
for s in shards:
p = folder / s["file"]
if not p.exists():
fails.append({"check": "missing", "file": s["file"], "text": "shard in the plan has no file"})
continue
warns += broken_links(p)
if prefix and s["file"].startswith(prefix):
fails.append(
{
"check": "name",
"file": s["file"],
"text": f"name starts with '{prefix}', so a '{prefix}*' glob would load it as another document",
}
)
sl = read_lines(p)
fm = frontmatter_end(sl)
if any(re.match(r"^(type|ticket)\s*:", x) for x in sl[:fm]):
fails.append(
{
"check": "frontmatter",
"file": s["file"],
"text": "shards carry no type or ticket keys; those stay in the main file",
}
)
source_path = folder / plan["source"]
source = read_lines(source_path) if source_path.exists() else None
if source is None:
fails.append({"check": "source", "file": plan["source"], "text": "the frozen original is missing"})
if rebuild and source is not None:
fm = plan.get("frontmatter_lines", 0)
joined = source[:fm]
for s in shards:
if (folder / s["file"]).exists():
joined += read_lines(folder / s["file"])[s.get("header_lines", 2) :]
if joined != source:
first = next(
(i for i, (x, y) in enumerate(zip(joined, source, strict=False), 1) if x != y),
min(len(joined), len(source)) + 1,
)
fails.append(
{
"check": "rebuild",
"file": plan["source"],
"line": first,
"text": "the shards do not join back into the source",
}
)
for s in shards:
p = folder / s["file"]
if p.exists() and (size := len(p.read_text(encoding="utf-8"))) > MAX_CHARS:
warns.append(
{
"check": "size",
"file": s["file"],
"text": f"{size} characters, over {MAX_CHARS}; consider splitting it",
}
)
if plan.get("lookup"):
pattern = re.compile(plan["lookup"])
homes: dict[str, set] = {}
for s in shards:
p = folder / s["file"]
if p.exists():
for line in body_lines(p):
if defines(line):
for ident in identifiers(pattern, line):
homes.setdefault(ident, set()).add(s["file"])
for ident, fs in sorted(homes.items()):
if len(fs) > 1:
warns.append(
{
"check": "identifier",
"file": ", ".join(sorted(fs)),
"text": f"{ident} is defined in more than one shard",
}
)
return fails, warns
def cmd_check(args) -> int:
target = Path(args.target).resolve()
if target.is_dir():
fails, warns = check_folder(target, args.rebuild)
elif target.is_file():
fails, warns = check_toc(target)
else:
return fail(f"not found: {target}")
return emit({"ok": not fails, "failures": fails, "warnings": warns}, 0 if not fails else 1)
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = ap.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("outline", help="measure a text file and propose a cut")
p.add_argument("source")
p.add_argument("--depth", type=int, default=3, help="list headings to this level (default 3)")
p.add_argument("--from", dest="from_line", type=int, help="list every heading from this line")
p.add_argument("--to", dest="to_line", type=int, help="list every heading up to this line")
p.add_argument("--ids", action="store_true", help="include each identifier defined in a heading and its line")
p = sub.add_parser(
"build",
help="build the sharded folder from a plan",
description=PLAN_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
p.add_argument("plan")
p = sub.add_parser("map", help="rewrite the main file's map from .shards.json")
p.add_argument("folder")
p = sub.add_parser("lookup", help="regenerate the identifier lookup")
p.add_argument("folder")
p = sub.add_parser(
"toc",
help="insert or refresh a contents list",
description=PLAN_HELP,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
p.add_argument("file")
p.add_argument("--plan", help="create the list from a plan (only groups, start and title are read)")
p.add_argument("--add", type=int, help="add a section starting at this line of the file as it reads now")
p.add_argument("--title", help="title of the section added with --add")
p.add_argument("--group", help="group of the section added with --add (default: the group before it)")
p.add_argument("--remove", action="store_true", help="remove the contents list and its record")
p = sub.add_parser("check", help="check a sharded folder or a file with a contents list")
p.add_argument("target")
p.add_argument("--rebuild", action="store_true", help="also check the faithful shards join back into the source")
args = ap.parse_args(argv)
try:
return {
"outline": cmd_outline,
"build": cmd_build,
"map": cmd_map,
"lookup": cmd_lookup,
"toc": cmd_toc,
"check": cmd_check,
}[args.cmd](args)
except Exception as e: # any crash is an input problem to report, not a check failure
return fail(f"{type(e).__name__}: {e}")
if __name__ == "__main__":
sys.exit(main())
scripts/tests/test_shard_tool.py
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# ///
"""Tests for shard_tool.py."""
import io
import json
import sys
from contextlib import redirect_stderr, redirect_stdout
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import shard_tool # noqa: E402
def run(*argv: str) -> tuple[int, dict]:
out = io.StringIO()
with redirect_stdout(out), redirect_stderr(io.StringIO()):
code = shard_tool.main(list(argv))
return code, json.loads(out.getvalue())
def doc(sections: int = 6, para: int = 40) -> str:
"""A markdown document with a title and sections of predictable size, one holding a table and a diagram."""
lines = ["# Sample", ""]
for i in range(1, sections + 1):
lines += [f"## Section {i}: FR-AB-{i:03d}", ""]
lines += [f"Line {i}.{j} with some words to fill it out for the size test." for j in range(para)]
if i == 2:
lines += [
"",
"| a | b |",
"|---|---|",
"| 1 | 2 |",
"",
"```mermaid",
"graph TD",
"## not a heading",
"```",
]
lines.append("")
return "\n".join(lines) + "\n"
def faithful_plan(tmp: Path, src: Path, lines: int, cut: list[int], extra: dict | None = None) -> Path:
"""A faithful plan cutting src at the given start lines."""
starts = [1] + cut
shards = []
for i, a in enumerate(starts):
b = starts[i + 1] - 1 if i + 1 < len(starts) else lines
shards.append(
{"file": f"part-{i + 1}.md", "title": f"Part {i + 1}", "summary": f"Lines {a} to {b}", "start": a, "end": b}
)
plan = {
"source": str(src),
"folder": str(tmp / "doc-sample"),
"title": "Sample",
"groups": [{"heading": "All", "shards": shards}],
}
plan |= extra or {}
path = tmp / "plan.json"
path.write_text(json.dumps(plan))
return path
@pytest.fixture
def built(tmp_path):
src = tmp_path / "sample.md"
src.write_text(doc())
lines = src.read_text().splitlines()
heads = [i for i, x in enumerate(lines, 1) if x.startswith("## Section")]
plan = faithful_plan(tmp_path, src, len(lines), heads[1:4], {"lookup": r"FR-AB-\d+"})
code, out = run("build", str(plan))
assert code == 0, out
return tmp_path / "doc-sample"
def test_outline_cut_respects_blocks_and_sizes(tmp_path):
src = tmp_path / "big.md"
src.write_text(doc(sections=12, para=120))
code, out = run("outline", str(src))
assert code == 0
assert all(h["text"] != "not a heading" for h in out["headings"])
blocks = [(b["start"], b["end"]) for b in out["blocks"]]
for seg in out["proposed_cut"]:
assert not any(a < seg["start"] <= b for a, b in blocks)
assert max(s["chars"] for s in out["proposed_cut"]) <= shard_tool.MAX_CHARS
starts = [s["start"] for s in out["proposed_cut"]]
assert starts == sorted(starts) and out["proposed_cut"][-1]["end"] == out["lines"]
def test_outline_flags_bare_identifier_headings(tmp_path):
src = tmp_path / "manual.md"
src.write_text("# Manual\n\n## 1\n\n### 1-1\n\ntext\n\n### C-2\n\ntext\n")
_, out = run("outline", str(src))
assert out["bare_heading_share"] == 1.0
def test_build_moves_source_and_rebuilds_exactly(built):
assert not (built.parent / "sample.md").exists()
assert (built / ".source" / "sample.md").exists()
main = (built / "doc-sample.md").read_text()
assert shard_tool.NOTE_MARK in main and "](./part-3.md)" in main and "identifier-lookup.md" in main
code, out = run("check", str(built), "--rebuild")
assert code == 0, out
def test_build_rejects_gaps(tmp_path):
src = tmp_path / "s.md"
src.write_text(doc())
plan = json.loads(faithful_plan(tmp_path, src, 10, [40]).read_text())
plan["groups"][0]["shards"][1]["start"] = 45
(tmp_path / "plan.json").write_text(json.dumps(plan))
code, out = run("build", str(tmp_path / "plan.json"))
assert code == 2 and "gap or overlap" in out["error"]
def test_build_keeps_frontmatter_in_main(tmp_path):
src = tmp_path / "s.md"
src.write_text("---\nstatus: draft\ntype: prd\n---\n" + doc())
n = len(src.read_text().splitlines())
run("build", str(faithful_plan(tmp_path, src, n, [60])))
folder = tmp_path / "doc-sample"
assert (folder / "doc-sample.md").read_text().startswith("---\nstatus: draft\ntype: prd\n---\n# Sample")
assert not (folder / "part-1.md").read_text().count("type: prd")
code, out = run("check", str(folder), "--rebuild")
assert code == 0, out
def test_long_map_splits_into_sub_maps(tmp_path):
src = tmp_path / "s.md"
src.write_text("\n".join(f"line {i}" for i in range(1, 501)) + "\n")
shards = [
{"file": f"s{i}.md", "title": f"S{i}", "summary": "x", "start": i * 2 + 1, "end": i * 2 + 2} for i in range(250)
]
groups = [{"heading": f"Group {g}", "shards": shards[g * 50 : (g + 1) * 50]} for g in range(5)]
plan = {"source": str(src), "folder": str(tmp_path / "big"), "title": "Big", "groups": groups}
(tmp_path / "p.json").write_text(json.dumps(plan))
code, out = run("build", str(tmp_path / "p.json"))
assert code == 0 and len(out["sub_maps"]) == 5
assert len((tmp_path / "big" / "big.md").read_text().splitlines()) <= shard_tool.WINDOW
code, out = run("check", str(tmp_path / "big"), "--rebuild")
assert code == 0, out
def test_check_catches_broken_links_orphans_names_and_edits(built):
(built / "part-2.md").write_text((built / "part-2.md").read_text() + "\nSee [gone](./gone.md).\n")
main = built / "doc-sample.md"
main.write_text("".join(x for x in main.read_text().splitlines(keepends=True) if "](./part-4.md)" not in x))
code, out = run("check", str(built), "--rebuild")
kinds = {f["check"] for f in out["failures"]}
assert code == 1 and {"link", "orphan", "rebuild"} <= kinds
plan = json.loads((built / ".shards.json").read_text())
plan["groups"][0]["shards"][0]["file"] = "doc-intro.md"
(built / "part-1.md").rename(built / "doc-intro.md")
(built / ".shards.json").write_text(json.dumps(plan))
_, out = run("check", str(built))
assert any(f["check"] == "name" for f in out["failures"])
def test_map_rewrites_after_a_new_shard(built):
(built / "extra.md").write_text("# Extra\n\nNew content.\n")
plan = json.loads((built / ".shards.json").read_text())
plan["groups"][0]["shards"].append({"file": "extra.md", "title": "Extra", "summary": "Added later"})
(built / ".shards.json").write_text(json.dumps(plan))
_, out = run("check", str(built))
assert any(f["check"] == "orphan" for f in out["failures"])
code, _ = run("map", str(built))
assert code == 0 and "](./extra.md)" in (built / "doc-sample.md").read_text()
code, out = run("check", str(built))
assert code == 0, out
def test_lookup_lists_each_identifier_home(built):
text = (built / "identifier-lookup.md").read_text()
assert "| FR-AB-001 | [part-1](./part-1.md) |" in text
def test_toc_insert_check_and_refresh(tmp_path):
f = tmp_path / "doc.md"
f.write_text(doc())
_, outline = run("outline", str(f))
starts = [h["line"] for h in outline["headings"] if h["level"] == 2]
plan = {
"groups": [{"heading": "All", "shards": [{"start": s, "title": f"Title {k}"} for k, s in enumerate(starts, 1)]}]
}
(tmp_path / "toc.json").write_text(json.dumps(plan))
code, out = run("toc", str(f), "--plan", str(tmp_path / "toc.json"))
assert code == 0 and out["fits_window"]
lines = f.read_text().splitlines()
entry = next(x for x in lines if x.endswith(" Title 3"))
start = int(entry.split()[1][1:].split("-")[0])
assert lines[start - 1] == "## Section 3: FR-AB-003"
assert run("check", str(f))[0] == 0
lines.insert(lines.index("## Section 2: FR-AB-002"), "Inserted paragraph.")
f.write_text("\n".join(lines) + "\n")
assert run("check", str(f))[0] == 1
run("toc", str(f))
assert run("check", str(f))[0] == 0
run("toc", str(f))
assert f.read_text().count(shard_tool.TOC_START) == 1
_, out = run("outline", str(f))
assert out["existing_contents"]["made_by"] == "shard-pro"
assert all(h["text"] != "Contents" for h in out["headings"])
def test_build_again_with_a_new_plan_replaces_old_shards(built):
plan = json.loads((built / ".shards.json").read_text())
shards = plan["groups"][0]["shards"]
merged = dict(shards[0], file="merged.md", title="Merged", end=shards[1]["end"])
plan["groups"][0]["shards"] = [merged] + shards[2:]
plan["source"] = "sample.md"
(built.parent / "plan2.json").write_text(json.dumps(plan))
code, out = run("build", str(built.parent / "plan2.json"))
assert code == 0, out
assert not (built / "part-1.md").exists() and (built / "merged.md").exists()
code, out = run("check", str(built), "--rebuild")
assert code == 0, out
def test_identifier_in_a_title_is_not_a_definition(tmp_path):
src = tmp_path / "s.md"
src.write_text("## FR-X-001 Thing\n\ntext\n\n## Other\n\nmore\n")
shards = [
{"file": "a.md", "title": "About FR-X-001", "summary": "a", "start": 1, "end": 4},
{"file": "b.md", "title": "FR-X-001 again", "summary": "b", "start": 5, "end": 7},
]
plan = {
"source": str(src),
"folder": str(tmp_path / "d"),
"title": "D",
"lookup": r"FR-X-\d+",
"groups": [{"heading": "G", "shards": shards}],
}
(tmp_path / "p.json").write_text(json.dumps(plan))
code, out = run("build", str(tmp_path / "p.json"))
assert out["lookup"] == {"identifiers": 1, "in_several_shards": []}
def test_outline_cuts_text_without_headings_by_size(tmp_path):
src = tmp_path / "t.txt"
src.write_text("".join(f"[00:{i // 60:02d}:{i % 60:02d}] Ana: {'word ' * 30}\n" for i in range(600)))
_, out = run("outline", str(src))
cut = out["proposed_cut"]
assert len(cut) > 5 and all(c["chars"] <= shard_tool.MAX_CHARS for c in cut)
assert cut[1]["heading"].startswith("[00:")
def test_check_accepts_a_relative_path(built, monkeypatch):
monkeypatch.chdir(built)
code, out = run("check", ".")
assert code == 0, out
def test_build_refuses_a_cut_inside_a_table_or_steps(tmp_path):
src = tmp_path / "s.md"
src.write_text("# T\n\n| a | b |\n|---|---|\n| 1 | 2 |\n| 3 | 4 |\n\n1. one\n2. two\n3. three\n")
for cut in (5, 9):
plan = faithful_plan(tmp_path, src, 10, [cut])
code, out = run("build", str(plan))
assert code == 2 and "inside the" in out["error"]
def test_bad_lookup_pattern_fails_before_anything_moves(tmp_path):
src = tmp_path / "s.md"
src.write_text(doc())
n = len(src.read_text().splitlines())
code, out = run("build", str(faithful_plan(tmp_path, src, n, [40], {"lookup": "FR-(["})))
assert code == 2 and "lookup pattern" in out["error"] and src.exists()
def test_toc_rejects_a_bad_plan_and_adds_a_section(tmp_path):
f = tmp_path / "doc.md"
f.write_text("# Doc\n\n## A\n\ntext a\n\n## B\n\ntext b\n\n## C\n\ntext c\n")
bad = {"groups": [{"heading": "G", "shards": [{"start": 11, "title": "C"}, {"start": 3, "title": "A"}]}]}
(tmp_path / "bad.json").write_text(json.dumps(bad))
code, out = run("toc", str(f), "--plan", str(tmp_path / "bad.json"))
assert code == 2 and "must rise" in out["error"]
good = {"groups": [{"heading": "G", "shards": [{"start": 3, "title": "A"}, {"start": 11, "title": "C"}]}]}
(tmp_path / "good.json").write_text(json.dumps(good))
assert run("toc", str(f), "--plan", str(tmp_path / "good.json"))[0] == 0
now = f.read_text().splitlines().index("## B") + 1
code, out = run("toc", str(f), "--add", str(now), "--title", "B")
assert code == 0 and out["entries"] == 3
lines = f.read_text().splitlines()
entry = next(x for x in lines if x.endswith(" B"))
assert lines[int(entry.split()[1][1:].split("-")[0]) - 1] == "## B"
assert run("check", str(f))[0] == 0
_, outline = run("outline", str(f))
assert outline["existing_contents"]["made_by"] == "shard-pro"
def test_oversized_shard_warns(built):
(built / "part-2.md").write_text("# Part 2\n\n" + "x" * (shard_tool.MAX_CHARS + 10) + "\n")
_, out = run("check", str(built))
assert any(w["check"] == "size" and w["file"] == "part-2.md" for w in out["warnings"])
def test_toc_plan_counts_lines_as_the_file_reads_now(tmp_path):
f = tmp_path / "doc.md"
f.write_text("# Doc\n\n## A\n\ntext a\n\n## B\n\ntext b\n")
plan = {"groups": [{"heading": "G", "shards": [{"start": 3, "title": "A"}, {"start": 7, "title": "B"}]}]}
(tmp_path / "p.json").write_text(json.dumps(plan))
run("toc", str(f), "--plan", str(tmp_path / "p.json"))
_, out = run("outline", str(f))
starts = [h["line"] for h in out["headings"] if h["level"] == 2]
plan = {
"groups": [
{"heading": "G", "shards": [{"start": starts[0], "title": "A2"}, {"start": starts[1], "title": "B2"}]}
]
}
(tmp_path / "p2.json").write_text(json.dumps(plan))
assert run("toc", str(f), "--plan", str(tmp_path / "p2.json"))[0] == 0
lines = f.read_text().splitlines()
entry = next(x for x in lines if x.endswith(" B2"))
assert lines[int(entry.split()[1][1:].split("-")[0]) - 1] == "## B"
assert run("check", str(f))[0] == 0
def test_toc_reports_a_renamed_section_and_keeps_the_rest(tmp_path):
f = tmp_path / "doc.md"
f.write_text("# Doc\n\n## A\n\ntext a\n\n## B\n\ntext b\n")
plan = {"groups": [{"heading": "G", "shards": [{"start": 3, "title": "A"}, {"start": 7, "title": "B"}]}]}
(tmp_path / "p.json").write_text(json.dumps(plan))
run("toc", str(f), "--plan", str(tmp_path / "p.json"))
f.write_text(f.read_text().replace("## B", "## B renamed"))
code, out = run("toc", str(f))
assert code == 1 and out["lost"][0]["title"] == "B" and out["entries"] == 1
now = f.read_text().splitlines().index("## B renamed") + 1
assert run("toc", str(f), "--add", str(now), "--title", "B")[0] == 0
assert run("check", str(f))[0] == 0
def test_check_toc_catches_a_stale_last_range(tmp_path):
f = tmp_path / "doc.md"
f.write_text("# Doc\n\n## A\n\ntext a\n")
plan = {"groups": [{"heading": "G", "shards": [{"start": 3, "title": "A"}]}]}
(tmp_path / "p.json").write_text(json.dumps(plan))
run("toc", str(f), "--plan", str(tmp_path / "p.json"))
f.write_text(f.read_text() + "more\nlines\n")
code, out = run("check", str(f))
assert code == 1 and "end of the file" in out["failures"][0]["text"]
def test_build_refuses_a_file_with_a_contents_list_until_removed(tmp_path):
src = tmp_path / "s.md"
src.write_text(doc())
lines = src.read_text().splitlines()
heads = [i for i, x in enumerate(lines, 1) if x.startswith("## ")]
toc = {"groups": [{"heading": "G", "shards": [{"start": h, "title": f"T{h}"} for h in heads]}]}
(tmp_path / "toc.json").write_text(json.dumps(toc))
run("toc", str(src), "--plan", str(tmp_path / "toc.json"))
code, out = run("build", str(faithful_plan(tmp_path, src, len(lines), heads[1:2])))
assert code == 2 and "--remove" in out["error"]
assert run("toc", str(src), "--remove")[0] == 0
assert src.read_text() == doc() and not (tmp_path / ".s.contents.json").exists()
assert run("build", str(faithful_plan(tmp_path, src, len(lines), heads[1:2])))[0] == 0
def test_loose_numbered_list_is_not_cut(tmp_path):
src = tmp_path / "s.md"
src.write_text("# T\n\n1. one\n\n2. two\n\n3. three\n")
code, out = run("build", str(faithful_plan(tmp_path, src, 7, [7])))
assert code == 2 and "steps" in out["error"]
def test_plan_missing_a_group_heading_fails_before_writing(tmp_path):
src = tmp_path / "s.md"
src.write_text("a\nb\n")
plan = {
"source": str(src),
"folder": str(tmp_path / "d"),
"title": "D",
"groups": [{"shards": [{"file": "a.md", "title": "A", "summary": "a", "start": 1, "end": 2}]}],
}
(tmp_path / "p.json").write_text(json.dumps(plan))
code, out = run("build", str(tmp_path / "p.json"))
assert code == 2 and "heading" in out["error"] and not (tmp_path / "d").exists()
def test_rebuild_in_place_reads_the_frozen_copy_not_the_map(tmp_path):
folder = tmp_path / "doc-sample"
folder.mkdir()
src = folder / "doc-sample.md"
src.write_text(doc())
n = len(src.read_text().splitlines())
plan = faithful_plan(tmp_path, src, n, [60])
assert run("build", str(plan))[0] == 0
code, out = run("build", str(plan))
assert code == 0, out
assert run("check", str(folder), "--rebuild")[0] == 0
def test_build_never_overwrites_a_file_it_did_not_write(tmp_path):
folder = tmp_path / "doc-sample"
folder.mkdir()
(folder / "part-2.md").write_text("someone else's notes\n")
src = tmp_path / "sample.md"
src.write_text(doc())
code, out = run("build", str(faithful_plan(tmp_path, src, len(doc().splitlines()), [60])))
assert code == 2 and "part-2.md already exists" in out["error"]
assert (folder / "part-2.md").read_text() == "someone else's notes\n" and src.exists()
def test_a_shard_cannot_take_a_name_build_writes_itself(tmp_path):
src = tmp_path / "sample.md"
src.write_text(doc())
plan = json.loads(faithful_plan(tmp_path, src, len(doc().splitlines()), [60], {"lookup": "FR"}).read_text())
plan["groups"][0]["shards"][1]["file"] = "identifier-lookup.md"
(tmp_path / "plan.json").write_text(json.dumps(plan))
code, out = run("build", str(tmp_path / "plan.json"))
assert code == 2 and "collide" in out["error"]
def test_plan_must_start_at_the_top_and_hold_shards(tmp_path):
src = tmp_path / "s.md"
src.write_text(doc())
n = len(doc().splitlines())
plan = json.loads(faithful_plan(tmp_path, src, n, [60]).read_text())
plan["groups"][0]["shards"] = plan["groups"][0]["shards"][1:]
(tmp_path / "plan.json").write_text(json.dumps(plan))
code, out = run("build", str(tmp_path / "plan.json"))
assert code == 2 and "first shard starts at line 1" in out["error"]
plan["groups"][0]["shards"] = []
(tmp_path / "plan.json").write_text(json.dumps(plan))
code, out = run("build", str(tmp_path / "plan.json"))
assert code == 2 and "no shards" in out["error"]
def test_a_group_too_big_for_one_map_page_is_refused(tmp_path):
src = tmp_path / "s.md"
src.write_text("\n".join(f"line {i}" for i in range(1, 301)) + "\n")
plan = faithful_plan(tmp_path, src, 300, list(range(2, 201)))
code, out = run("build", str(plan))
assert code == 2 and "map page holds" in out["error"]
def test_check_warns_about_links_the_split_moved(tmp_path):
(tmp_path / "guide.md").write_text("# Guide\n")
src = tmp_path / "sample.md"
text = doc().replace("Line 1.0 with", "See [guide](guide.md) and [the end](#section-6-fr-ab-006). Line 1.0 with")
src.write_text(text)
lines = text.splitlines()
heads = [i for i, x in enumerate(lines, 1) if x.startswith("## Section")]
assert run("build", str(faithful_plan(tmp_path, src, len(lines), heads[1:])))[0] == 0
code, out = run("check", str(tmp_path / "doc-sample"), "--rebuild")
assert code == 0
moved = [w["text"] for w in out["warnings"] if w["check"] == "link"]
assert any("guide.md" in t for t in moved) and any("#section-6" in t for t in moved)
def test_toc_refresh_refuses_to_guess_between_two_matching_sections(tmp_path):
f = tmp_path / "doc.md"
f.write_text(doc())
_, outline = run("outline", str(f))
starts = [h["line"] for h in outline["headings"] if h["level"] == 2]
plan = {"groups": [{"heading": "All", "shards": [{"start": s, "title": f"T{k}"} for k, s in enumerate(starts)]}]}
(tmp_path / "toc.json").write_text(json.dumps(plan))
run("toc", str(f), "--plan", str(tmp_path / "toc.json"))
lines = f.read_text().splitlines()
lines.insert(lines.index("## Section 2: FR-AB-002"), "## Section 5: FR-AB-005")
f.write_text("\n".join(lines) + "\n")
before = f.read_text()
code, out = run("toc", str(f))
assert code == 1 and out["ambiguous"][0]["first_line"] == "## Section 5: FR-AB-005"
assert f.read_text() == before
assert run("check", str(f))[0] == 1
def test_a_longer_fence_holds_a_shorter_one(tmp_path):
lines = ["# T", "", "````markdown", "```", "## not a heading", "```", "````", "", "## Real"]
spans = shard_tool.block_spans(lines)
assert spans[0] == (3, 7, "markdown")
heads = shard_tool.headings(lines, spans)
assert [h["text"] for h in heads] == ["T", "Real"]
SKILL.md
---
name: bmad-shard-pro
description: 'Makes long documents easy for agents to navigate. Use when the user says "shard this document", "this file is too long for agents", "turn this document into a wiki", "add a table of contents so agents can find things", or "update the sharded document".'
---
# Shard Pro
You make a long document cheap and safe for an agent to read later. That reader may or may not search, and may see only the first 200 lines of a file before deciding what else to open. From what it sees there, it must be able to tell where any fact lives. Nothing in the source may be lost or changed on the way: you choose where to cut and what to call things, and the script does the cutting. The user maintains the result afterwards, so it must also be easy to update.
## Conventions
- `{skill-root}` is this skill's folder; `{project-root}` is the nearest folder containing `_bmad/`.
- This skill's script is `uv run {skill-root}/scripts/shard_tool.py`; run it with `--help`, or `<command> --help`, for arguments. The memlog script is `uv run {project-root}/_bmad/scripts/memlog.py`; without `_bmad/`, skip memlog writes and say so once.
- `{doc_folder}` is the folder that holds the result. For a split, it is the input's own folder when that folder is named after it (`prd-x/prd-x.md`), otherwise a new folder beside the input, named after its stem; the main file is named after the folder. For a contents list, it is the file's own folder.
## On activation
- The document's main file opens with the sharded-document note, or the document has a contents list this skill made (it starts with `<!-- shard-pro:contents -->`): read `references/update.md` fully and follow it.
- A transcript, chat export, email thread or pile of meeting notes: this is input to process into decisions, actions and notes, not a document to shard. Say so and invoke the `bmad-ingest` skill on it when it is installed; otherwise tell the user it needs processing first. Then stop.
- Any other file that is not markdown or plain text (.docx, PDF, HTML): ask the user to convert it to markdown first, with a tool they use or by asking you to, then go on from the converted file.
- Any other document: go to Assess.
## Assess
Run `outline` on the input. Recommend one outcome with one line on why:
- **Leave it as it is.** It is short enough to read whole, or already well signposted.
- **Contents list.** The file stays one file and gains a detailed contents list near the top. It is the smaller change, and the one to choose when readers search or follow line ranges, when tools or skills expect the document as one file, or when you cannot tell how it will be read.
- **Faithful split.** The source is cut word for word into shards with a map. Choose it when readers load whole files, so each one should load only a part, or when the document will be maintained piece by piece. It loses nothing and is checked by script.
Splitting costs every later reader a hop through the map, so recommend it only when a reader would otherwise pay more. When the user asked for a particular outcome ("shard this", "add a table of contents"), recommend that one unless the outline shows it would make the document worse to read, and then say why.
For Leave it as it is, tell the user why and stop. The other two go on to Plan.
## Plan from the outline
Plan from the outline and read only the ranges you need. Holding a long document whole in your context is the expensive mistake. Start from the proposed cut and change it where the content asks:
- Shards run about 3 to 14 KB, measured in characters, because line lengths vary too much to measure by lines. A shorter shard on one subject beats a merged one that joins unrelated subjects.
- Never cut inside a table, a step list or a diagram. `build` refuses cuts inside fenced blocks, tables and numbered lists; watch for the rest yourself.
- Name each shard from what it covers, in kebab-case. Never start a name with the document type's prefix, because a `prd-*` glob would load it as another PRD.
- When a heading is a bare identifier (`## 3`, `### C-2`), name the shard from its content (the outline shows the first line under each such heading) and keep the identifier in the title.
- Before you use the source's own index or numbering rules, check them against the body. They are often from an older edition.
- Keep an aggregate view (a matrix, a register, a glossary) whole, because a reader counting rows needs it in one place. Small ones may share a shard.
- Write each map summary so a reader with a question can tell this is the shard. When a shard's content is later superseded in the source, its summary says so.
- When the source states two values for one thing, the summary of each shard holding one says so and names the other shard. Never pick one, even where the source declares a precedence rule.
- Set a lookup pattern only when readers arrive with identifiers the map titles do not already resolve, such as hundreds of requirement IDs.
Write the plan as JSON in the shape `build --help` shows.
## Agree the plan
- A user is present: show the outcome, why, where the result will go, and the shard list (titles and one-line summaries), then wait. For a split, say that the input moves into `.source/` and the shards become the document. Build only once they agree.
- The user asked you to run without stopping, or another skill invoked this one with no user present: go on, and report at the end.
## Build
Run `build`, or `toc` with the plan for a contents list. To change the plan afterwards, edit it and run `build` again; it reads the source from `.source/` and replaces the old shards.
Then record the outcome and each change or decision the user made in `{doc_folder}/.memlog.md`, keeping any memlog already there: `memlog.py init --workspace {doc_folder}` when none exists, then `append --type decision|change|event`.
## Check, then report
Run `check` on the result, with `--rebuild` after a faithful split, and fix every failure before you finish.
Report to the user:
- the outcome, and where the main file is;
- for a split, that the input now sits in `.source/`;
- what you found in the source: conflicts with both values and where each sits, index errors, suspected errors;
- links `check` warns no longer resolve after the split. Shards keep the source's words, so the user decides whether to fix them.
Findings go in the report and the memlog, never into shards. A shard says what the source says, and a reader cannot tell your judgement from the author's.