fix(lint_roadmap): check Blocked-by fragments into YAML and JSON decision files

changelog.d/3567-the-roadmap-lint-checks-a-blocked-by-fra.md

@@ -0,0 +1,5 @@
+---
+class: fixed
+trace: #3567
+---
+The roadmap lint checks a Blocked-by fragment into a YAML or JSON decision file when `decisions.entryKey` names the key holding each entry id; without the key, behavior is unchanged.

specfuse/loop/layout.py

@@ -6,7 +6,8 @@
 """Project layout resolver — FEAT-2026-0224 (project-configurable layout).
 
 Reads an optional `.specfuse/layout.yml` and answers `layout get <key>` for
-`decisions.style`, `decisions.location`, `decisions.link` and `ids.key`.
+`decisions.style`, `decisions.location`, `decisions.link`, `decisions.entryKey`
+and `ids.key`.
 
 Contract: no file (or no `decisions:` block) means today's behaviour — style
 `adr`, location `docs/adr`, link `{path}`. An `entries` style with `location`
@@ -41,7 +42,14 @@
 }
 _FALLBACK_DEFAULTS = _STYLE_DEFAULTS["adr"]
 
-KEYS = ("decisions.style", "decisions.location", "decisions.link", "ids.key", "changelog.mode")
+KEYS = (
+    "decisions.style",
+    "decisions.location",
+    "decisions.link",
+    "decisions.entryKey",
+    "ids.key",
+    "changelog.mode",
+)
 
 DEFAULT_CHANGELOG_MODE = "single"
 VALID_CHANGELOG_MODES = ("single", "fragments")
@@ -50,6 +58,7 @@
 LINK_PLACEHOLDERS = ("location", "path", "id", "domain")
 KNOWN_TOP_LEVEL = ("decisions", "ids", "changelog")
 _PLACEHOLDER_RE = re.compile(r"\{([^{}]*)\}")
+_ENTRY_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_-]*$")
 _DRIVE_RE = re.compile(r"^[A-Za-z]:")
 
 
@@ -58,6 +67,7 @@ class Decisions:
     style: str
     location: str
     link: str
+    entry_key: Optional[str] = None
 
 
 @dataclass(frozen=True)
@@ -90,10 +100,12 @@ def _resolve_decisions(block: Optional[dict]) -> Decisions:
     default_location, default_link = _STYLE_DEFAULTS.get(style, _FALLBACK_DEFAULTS)
     location = block.get("location")
     link = block.get("link")
+    entry_key = block.get("entryKey")
     return Decisions(
         style=style,
         location=default_location if location is None else _normalise(str(location)),
         link=default_link if link is None else str(link),
+        entry_key=None if entry_key is None else str(entry_key),
     )
 
 
@@ -250,6 +262,18 @@ def add(severity, key, message):
                     f"decisions.link placeholder '{{{name}}}' is not allowed; "
                     "use {location}, {path}, {id} or {domain}",
                 )
+
+    entry_key = block.get("entryKey")
+    if entry_key is not None:
+        if not _ENTRY_KEY_RE.match(str(entry_key)):
+            add(
+                SEVERITY_ERROR,
+                "entryKey",
+                f"decisions.entryKey {str(entry_key)!r} is invalid; must match "
+                f"{_ENTRY_KEY_RE.pattern}",
+            )
+        if style is None or str(style) != "entries":
+            add(SEVERITY_ERROR, "entryKey", "decisions.entryKey applies only to style 'entries'")
     return findings
 
 
@@ -264,6 +288,8 @@ def get(layout: Layout, key: str) -> str:
         return layout.ids.key or ""
     if key == "changelog.mode":
         return layout.changelog.mode
+    if key == "decisions.entryKey":
+        return layout.decisions.entry_key or ""
     return getattr(layout.decisions, key.split(".", 1)[1])
 
 

specfuse/loop/lint_roadmap.py

@@ -44,6 +44,7 @@
 from dataclasses import dataclass
 from pathlib import Path
 from urllib.parse import urlparse
+import json
 import re
 
 from specfuse.loop import ids as _ids
@@ -160,6 +161,7 @@ def _find_anchors(lines):
     return anchors
 
 
+_ENTRY_FILE_SUFFIXES = (".yaml", ".yml", ".json")
 _EXPLICIT_ANCHOR_RE = re.compile(r'<a\s+(?:id|name)\s*=\s*"([^"]+)"', re.IGNORECASE)
 _MD_HEADING_RE = re.compile(r'^ {0,3}#{1,6}\s+(.*?)\s*#*\s*$')
 
@@ -193,6 +195,58 @@ def _collect_anchor_names(text: str) -> set:
     return names
 
 
+def _json_entry_ids(value, entry_key: str) -> set:
+    """Every scalar `entry_key` value in a parsed JSON document, at any depth."""
+    ids = set()
+    if isinstance(value, dict):
+        for key, item in value.items():
+            if key == entry_key and isinstance(item, (str, int)) and not isinstance(item, bool):
+                ids.add(str(item))
+            ids |= _json_entry_ids(item, entry_key)
+    elif isinstance(value, list):
+        for item in value:
+            ids |= _json_entry_ids(item, entry_key)
+    return ids
+
+
+def _collect_entry_ids(path: Path, entry_key: str):
+    """Ids a YAML or JSON decision file answers to, or None when it cannot be read (#3567).
+
+    YAML is scanned by line rather than parsed: `_miniyaml` refuses block scalars
+    and flow mappings by design, and decision files hold long rationale. A line
+    inside a block scalar that looks like the key therefore counts as an id, which
+    can only let a typo pass, never fail a good link.
+    """
+    try:
+        text = path.read_text(encoding="utf-8")
+        if path.suffix == ".json":
+            return _json_entry_ids(json.loads(text), entry_key)
+    except (OSError, UnicodeDecodeError, ValueError):
+        return None
+    line_re = re.compile(
+        rf"""^\s*(?:-\s+)*["']?{re.escape(entry_key)}["']?\s*:\s+"""
+        r"""(?:"([^"]*)"|'([^']*)'|([^\s#"'][^#]*?))\s*(?:\#.*)?$"""
+    )
+    ids = set()
+    for line in text.splitlines():
+        m = line_re.match(line)
+        if m:
+            ids.add(next(group for group in m.groups() if group is not None))
+    return ids
+
+
+def _entry_anchor_problem(target: Path, link_path: str, fragment: str, entry_key: str):
+    """Why `fragment` does not resolve in a YAML or JSON decision file, or None."""
+    ids = _collect_entry_ids(target, entry_key)
+    if ids is None:
+        return f"'{link_path}', which could not be read as YAML or JSON"
+    if not ids:
+        return f"'{link_path}', which has no entries under entryKey '{entry_key}'"
+    if fragment not in ids:
+        return f"'{link_path}', which has no anchor '{fragment}'"
+    return None
+
+
 def _declared_decisions_location(repo_root: Path):
     """Resolved `decisions.location` only when layout.yml declares a `decisions:` block."""
     data = _layout._read(repo_root / _layout.LAYOUT_RELPATH)
@@ -419,6 +473,7 @@ def _find_blocked_by(section, lines):
 def _check_blocked_by(roadmap_doc: _FileDoc, archive_doc: _FileDoc, rows, repo_root: Path) -> list:
     findings = []
     declared_location = _declared_decisions_location(repo_root)
+    entry_key = _layout.load(repo_root).decisions.entry_key
     roadmap_sections = _detail_sections(roadmap_doc.lines)
     archive_sections = _detail_sections(archive_doc.lines)
 
@@ -488,6 +543,16 @@ def _check_blocked_by(roadmap_doc: _FileDoc, archive_doc: _FileDoc, rows, repo_r
                             f"{row['id']}'s Blocked-by link '{href}' points at "
                             f"'{link_path}', which has no anchor '{fragment}'",
                         ))
+                elif (
+                    fragment and entry_key and target.is_file()
+                    and target.suffix in _ENTRY_FILE_SUFFIXES
+                ):
+                    problem = _entry_anchor_problem(target, link_path, fragment, entry_key)
+                    if problem:
+                        findings.append(Finding(
+                            SEVERITY_ERROR, doc.filename, blocked_by_line + 1,
+                            f"{row['id']}'s Blocked-by link '{href}' points at {problem}",
+                        ))
                 if declared_location is not None:
                     location = (repo_root / declared_location).resolve()
                     try:

tests/test_blocked_by_entry_anchor.py

@@ -0,0 +1,163 @@
+#
+# Copyright 2026 Specfuse contributors
+# Licensed under the Apache License, Version 2.0. See LICENSE.
+#
+"""A Blocked-by fragment into a YAML or JSON decision file is checked (#3567).
+
+`decisions.entryKey` names the key holding an entry's id. Without it, a fragment
+into a non-Markdown file stays unchecked, as it was before the key existed.
+"""
+
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from specfuse.loop import layout
+from specfuse.loop.lint_roadmap import SEVERITY_ERROR, lint_roadmap
+
+LAYOUT = """decisions:
+  style: entries
+  location: decisions
+  link: "{location}/{domain}.yaml#{id}"
+"""
+KEYED_LAYOUT = LAYOUT + "  entryKey: name\n"
+
+DECISIONS = """entries:
+  - name: pricing-is-per-seat
+    why: "cost tracks usage"
+    tags: [billing, pricing]
+    limits: {seats: 5}
+    detail: |
+      Long rationale that the loop's own YAML parser refuses.
+  - "name": "quoted-id"  # a trailing comment
+  -   name: 'single-quoted-id'
+"""
+
+
+def _errors(root, href, *, layout_text=KEYED_LAYOUT, files=None):
+    root = Path(root)
+    (root / ".specfuse").mkdir()
+    (root / ".specfuse" / "layout.yml").write_text(layout_text)
+    (root / ".specfuse" / "roadmap.md").write_text(
+        "# Roadmap\n\n| Feature ID | Title | Status | Folder | Detail |\n|---|---|---|---|---|\n"
+        "| FEAT-2026-0001 | T | blocked | — | — |\n\n"
+        '<a id="feat-2026-0001"></a>\n## FEAT-2026-0001 — T\n\n'
+        f"**Blocked by.** [D]({href}) — awaiting signature.\n"
+    )
+    (root / "decisions").mkdir()
+    for name, text in (files or {"billing.yaml": DECISIONS}).items():
+        (root / "decisions" / name).write_text(text)
+    return [
+        f.message
+        for f in lint_roadmap(root)
+        if f.severity == SEVERITY_ERROR and "Blocked-by" in f.message
+    ]
+
+
+class YamlTargets(unittest.TestCase):
+    def _lint(self, fragment, **kwargs):
+        with tempfile.TemporaryDirectory() as tmp:
+            return _errors(tmp, f"../decisions/billing.yaml#{fragment}", **kwargs)
+
+    def test_a_typo_is_an_error_worded_like_markdown(self):
+        errors = self._lint("pricing-is-per-sat")
+        self.assertEqual(len(errors), 1, errors)
+        self.assertIn(
+            "points at '../decisions/billing.yaml', which has no anchor 'pricing-is-per-sat'",
+            errors[0],
+        )
+
+    def test_an_existing_id_passes_beside_syntax_the_loop_parser_rejects(self):
+        self.assertEqual(self._lint("pricing-is-per-seat"), [])
+
+    def test_quoted_key_and_values_with_a_comment_pass(self):
+        self.assertEqual(self._lint("quoted-id"), [])
+        self.assertEqual(self._lint("single-quoted-id"), [])
+
+    def test_yml_suffix_is_checked(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            errors = _errors(
+                tmp, "../decisions/billing.yml#nope", files={"billing.yml": DECISIONS}
+            )
+        self.assertEqual(len(errors), 1, errors)
+
+    def test_a_file_with_no_entries_under_the_key_names_the_key(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            errors = _errors(
+                tmp, "../decisions/billing.yaml#x", files={"billing.yaml": "id: x\n"}
+            )
+        self.assertEqual(len(errors), 1, errors)
+        self.assertIn("no entries under entryKey 'name'", errors[0])
+
+    def test_without_entry_key_a_typo_still_passes(self):
+        self.assertEqual(self._lint("pricing-is-per-sat", layout_text=LAYOUT), [])
+
+    def test_a_link_with_no_fragment_passes_for_a_decision_still_under_review(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self.assertEqual(_errors(tmp, "../decisions/billing.yaml"), [])
+
+
+class JsonTargets(unittest.TestCase):
+    TEXT = json.dumps({"entries": [{"name": "real", "nested": {"name": "inner"}}, {"name": 7}]})
+
+    def _lint(self, fragment, text=TEXT):
+        with tempfile.TemporaryDirectory() as tmp:
+            return _errors(
+                tmp, f"../decisions/billing.json#{fragment}", files={"billing.json": text}
+            )
+
+    def test_existing_ids_pass(self):
+        self.assertEqual(self._lint("real"), [])
+        self.assertEqual(self._lint("7"), [])
+
+    def test_a_typo_is_an_error(self):
+        errors = self._lint("reel")
+        self.assertEqual(len(errors), 1, errors)
+        self.assertIn("has no anchor 'reel'", errors[0])
+
+    def test_malformed_json_is_a_finding_not_a_silent_pass(self):
+        errors = self._lint("real", text="{not json")
+        self.assertEqual(len(errors), 1, errors)
+        self.assertIn("could not be read", errors[0])
+
+
+class MarkdownTargets(unittest.TestCase):
+    def test_markdown_still_answers_to_anchors_when_the_key_is_set(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            errors = _errors(
+                tmp, "../decisions/d.md#pricing", files={"d.md": "## Pricing\n"}
+            )
+        self.assertEqual(errors, [])
+
+
+class LayoutKey(unittest.TestCase):
+    def _write(self, tmp, text):
+        (Path(tmp) / ".specfuse").mkdir()
+        (Path(tmp) / ".specfuse" / "layout.yml").write_text(text)
+
+    def _errors(self, text):
+        with tempfile.TemporaryDirectory() as tmp:
+            self._write(tmp, text)
+            return [f.message for f in layout.validate(tmp) if f.severity == SEVERITY_ERROR]
+
+    def test_get_returns_the_key_and_empty_when_unset(self):
+        with tempfile.TemporaryDirectory() as tmp:
+            self.assertEqual(layout.get(layout.load(tmp), "decisions.entryKey"), "")
+            self._write(tmp, KEYED_LAYOUT)
+            self.assertEqual(layout.get(layout.load(tmp), "decisions.entryKey"), "name")
+
+    def test_a_valid_key_has_no_findings(self):
+        self.assertEqual(self._errors(KEYED_LAYOUT), [])
+
+    def test_a_key_that_cannot_name_a_mapping_key_is_an_error(self):
+        errors = self._errors(LAYOUT + '  entryKey: "not a key"\n')
+        self.assertTrue(any("decisions.entryKey" in m for m in errors), errors)
+
+    def test_the_key_is_refused_under_adr_style(self):
+        errors = self._errors("decisions:\n  style: adr\n  entryKey: name\n")
+        self.assertTrue(any("decisions.entryKey" in m for m in errors), errors)
+
+
+if __name__ == "__main__":
+    unittest.main()