refactor: split frontmatter parsing + VerifyOutcome out of verify.py (#242)

src/bmad_loop/escalation.py

@@ -12,9 +12,8 @@
 from typing import Any
 
 from .adapters.base import SessionResult
-from .model import StoryTask
+from .model import StoryTask, VerifyOutcome
 from .policy import Policy
-from .verify import VerifyOutcome
 
 SEVERITY_CRITICAL = "CRITICAL"
 SEVERITY_PREFERENCE = "PREFERENCE"

src/bmad_loop/frontmatter.py

@@ -0,0 +1,106 @@
+"""Pure spec-frontmatter parsing: read the YAML ``---``…``---`` block, normalize
+the status token, and rewrite ``status:`` in place.
+
+Zero git/subprocess dependencies (only stdlib + PyYAML) so pure domain modules
+(``stories``, ``devcontract``) can read spec status without importing ``verify``
+and dragging in its whole git surface (assessment finding F-1). ``verify``
+re-exports these names, so every existing ``verify.<name>`` / ``from .verify
+import <name>`` call site stays valid.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Any
+
+import yaml
+
+
+def _split_frontmatter(text: str) -> tuple[str, str, str] | None:
+    """Split a document into ``(before, block, after)`` around its YAML
+    frontmatter, where ``before + block + after == text`` exactly.
+
+    The opening and closing ``---`` are recognized ONLY as standalone delimiter
+    lines (``line.rstrip() == "---"``), so a ``---`` substring inside a scalar
+    value (e.g. ``title: 'restore --- review'``) is never mistaken for the
+    closing boundary — the flaw a plain ``text.split("---", 2)`` has. ``before``
+    is the opening delimiter line, ``block`` is the YAML content between the
+    delimiters, and ``after`` begins with the closing delimiter line; callers
+    rewrite ``block`` and reconstruct the file byte-for-byte. Returns ``None``
+    when the text has no opening delimiter line or no closing delimiter line.
+    """
+    lines = text.splitlines(keepends=True)  # "".join(lines) == text
+    if not lines or lines[0].rstrip() != "---":
+        return None
+    for i in range(1, len(lines)):
+        if lines[i].rstrip() == "---":
+            return lines[0], "".join(lines[1:i]), "".join(lines[i:])
+    return None
+
+
+def read_frontmatter(path: Path) -> dict[str, Any]:
+    if not path.is_file():
+        return {}
+    try:
+        text = path.read_text(encoding="utf-8")
+    except UnicodeDecodeError:
+        # A non-UTF-8 file carries no readable frontmatter — degrade exactly like
+        # unparseable YAML below. Every status gate then reads status "" and
+        # returns a clean retry/repair outcome instead of crashing mid-verify
+        # (UnicodeDecodeError is a ValueError, so it slipped past callers'
+        # except-OSError guards).
+        return {}
+    split = _split_frontmatter(text)
+    if split is None:
+        return {}
+    try:
+        doc = yaml.safe_load(split[1])
+    except yaml.YAMLError:
+        return {}
+    return doc if isinstance(doc, dict) else {}
+
+
+def status_of(fm: dict[str, Any]) -> str:
+    """Normalized spec status from a frontmatter dict: stripped + lowercased.
+
+    The single point all spec-frontmatter status gates read through, so casing
+    never decides a gate — the spec template and sprint-status tokens are
+    lowercase, so a stray ``Done``/``In-Review`` from a hand-edited spec still
+    matches. (``devcontract`` keeps its own lowercasing; it parses skill-written
+    prose where casing genuinely varies.)
+    """
+    return str(fm.get("status", "")).strip().lower()
+
+
+def set_frontmatter_status(path: Path, status: str) -> bool:
+    """Rewrite the `status:` field in a spec's `---`…`---` frontmatter block.
+
+    A minimal in-place line replacement (not a YAML round-trip) so the spec's
+    formatting, comments, and field order survive — only the status value
+    changes. Returns True when the file was rewritten, False when it has no
+    frontmatter or already carries `status`. Idempotent.
+    """
+    if not path.is_file():
+        return False
+    text = path.read_text(encoding="utf-8")
+    split = _split_frontmatter(text)
+    if split is None:
+        return False
+    before, block, after = split
+    block_lines = block.splitlines(keepends=True)
+    replaced = False
+    for i, line in enumerate(block_lines):
+        stripped = line.lstrip()
+        if stripped.startswith("status:") and not stripped.startswith("status_"):
+            indent = line[: len(line) - len(stripped)]
+            newline = "\n" if line.endswith("\n") else ""
+            block_lines[i] = f"{indent}status: {status}{newline}"
+            replaced = True
+            break
+    if not replaced:
+        return False
+    rebuilt = before + "".join(block_lines) + after
+    if rebuilt == text:  # already at the target value — idempotent no-op
+        return False
+    path.write_text(rebuilt, encoding="utf-8")
+    return True

src/bmad_loop/model.py

@@ -458,3 +458,35 @@ def from_dict(cls, d: dict[str, Any]) -> "RunState":
             plugin_shared=dict(d.get("plugin_shared", {})),
             tasks={k: StoryTask.from_dict(t) for k, t in d.get("tasks", {}).items()},
         )
+
+
+@dataclass(frozen=True)
+class VerifyOutcome:
+    ok: bool
+    reason: str = ""
+    severity: str = ""  # "" | "CRITICAL" | "PREFERENCE" — set when not retryable
+    # fixable failures carry concrete evidence (failing command output) that a
+    # feedback-driven repair session can act on; non-fixable retries start over
+    fixable: bool = False
+    # the failure is the run environment's, not the story's (verify command
+    # not found / not executable): no repair session can fix it and every
+    # story shares the same commands, so it must never charge attempt budgets
+    env_fault: bool = False
+
+    @classmethod
+    def passed(cls) -> "VerifyOutcome":
+        return cls(ok=True)
+
+    @classmethod
+    def retry(cls, reason: str, fixable: bool = False) -> "VerifyOutcome":
+        return cls(ok=False, reason=reason, fixable=fixable)
+
+    @classmethod
+    def escalate(
+        cls, reason: str, severity: str = "CRITICAL", env_fault: bool = False
+    ) -> "VerifyOutcome":
+        return cls(ok=False, reason=reason, severity=severity, env_fault=env_fault)
+
+    @property
+    def retryable(self) -> bool:
+        return not self.ok and not self.severity

src/bmad_loop/stories.py

@@ -28,7 +28,7 @@
 
 import yaml
 
-from .verify import read_frontmatter, status_of
+from .frontmatter import read_frontmatter, status_of
 
 # Fixed-name discovery, like SPEC.md / .memlog.md — never listed in companions.
 STORIES_FILENAME = "stories.yaml"

src/bmad_loop/verify.py

@@ -15,11 +15,11 @@
 from pathlib import Path
 from typing import Any
 
-import yaml
-
 from . import deferredwork
 from .bmadconfig import ProjectPaths
-from .model import StoryTask
+from .frontmatter import set_frontmatter_status  # noqa: F401 — re-export
+from .frontmatter import _split_frontmatter, read_frontmatter, status_of
+from .model import StoryTask, VerifyOutcome
 from .policy import POLICY_FILE, Policy
 from .sprintstatus import story_status
 
@@ -66,38 +66,6 @@ class GitError(Exception):
     pass
 
 
-@dataclass(frozen=True)
-class VerifyOutcome:
-    ok: bool
-    reason: str = ""
-    severity: str = ""  # "" | "CRITICAL" | "PREFERENCE" — set when not retryable
-    # fixable failures carry concrete evidence (failing command output) that a
-    # feedback-driven repair session can act on; non-fixable retries start over
-    fixable: bool = False
-    # the failure is the run environment's, not the story's (verify command
-    # not found / not executable): no repair session can fix it and every
-    # story shares the same commands, so it must never charge attempt budgets
-    env_fault: bool = False
-
-    @classmethod
-    def passed(cls) -> "VerifyOutcome":
-        return cls(ok=True)
-
-    @classmethod
-    def retry(cls, reason: str, fixable: bool = False) -> "VerifyOutcome":
-        return cls(ok=False, reason=reason, fixable=fixable)
-
-    @classmethod
-    def escalate(
-        cls, reason: str, severity: str = "CRITICAL", env_fault: bool = False
-    ) -> "VerifyOutcome":
-        return cls(ok=False, reason=reason, severity=severity, env_fault=env_fault)
-
-    @property
-    def retryable(self) -> bool:
-        return not self.ok and not self.severity
-
-
 def _run_git(
     cmd: list[str], repo: Path, *, env: dict[str, str] | None = None
 ) -> subprocess.CompletedProcess[str]:
@@ -935,77 +903,6 @@ def capture_diff(repo: Path, baseline: str, *, max_file_bytes: int | None = None
     return "".join(parts)
 
 
-def read_frontmatter(path: Path) -> dict[str, Any]:
-    if not path.is_file():
-        return {}
-    try:
-        text = path.read_text(encoding="utf-8")
-    except UnicodeDecodeError:
-        # A non-UTF-8 file carries no readable frontmatter — degrade exactly like
-        # unparseable YAML below. Every status gate then reads status "" and
-        # returns a clean retry/repair outcome instead of crashing mid-verify
-        # (UnicodeDecodeError is a ValueError, so it slipped past callers'
-        # except-OSError guards).
-        return {}
-    if not text.startswith("---"):
-        return {}
-    parts = text.split("---", 2)
-    if len(parts) < 3:
-        return {}
-    try:
-        doc = yaml.safe_load(parts[1])
-    except yaml.YAMLError:
-        return {}
-    return doc if isinstance(doc, dict) else {}
-
-
-def status_of(fm: dict[str, Any]) -> str:
-    """Normalized spec status from a frontmatter dict: stripped + lowercased.
-
-    The single point all spec-frontmatter status gates read through, so casing
-    never decides a gate — the spec template and sprint-status tokens are
-    lowercase, so a stray ``Done``/``In-Review`` from a hand-edited spec still
-    matches. (``devcontract`` keeps its own lowercasing; it parses skill-written
-    prose where casing genuinely varies.)
-    """
-    return str(fm.get("status", "")).strip().lower()
-
-
-def set_frontmatter_status(path: Path, status: str) -> bool:
-    """Rewrite the `status:` field in a spec's `---`…`---` frontmatter block.
-
-    A minimal in-place line replacement (not a YAML round-trip) so the spec's
-    formatting, comments, and field order survive — only the status value
-    changes. Returns True when the file was rewritten, False when it has no
-    frontmatter or already carries `status`. Idempotent.
-    """
-    if not path.is_file():
-        return False
-    text = path.read_text(encoding="utf-8")
-    if not text.startswith("---"):
-        return False
-    parts = text.split("---", 2)
-    if len(parts) < 3:
-        return False
-    block_lines = parts[1].splitlines(keepends=True)
-    replaced = False
-    for i, line in enumerate(block_lines):
-        stripped = line.lstrip()
-        if stripped.startswith("status:") and not stripped.startswith("status_"):
-            indent = line[: len(line) - len(stripped)]
-            newline = "\n" if line.endswith("\n") else ""
-            block_lines[i] = f"{indent}status: {status}{newline}"
-            replaced = True
-            break
-    if not replaced:
-        return False
-    rebuilt = parts[0] + "---" + "".join(block_lines) + "---" + parts[2]
-    if rebuilt == text:  # already at the target value — idempotent no-op
-        return False
-    path.write_text(rebuilt, encoding="utf-8")
-    return True
-
-
 def set_frontmatter_field(path: Path, key: str, value: str) -> bool:
     """Rewrite (or insert) a scalar ``<key>:`` line in a spec's `---`…`---`
     frontmatter block.
@@ -1021,12 +918,11 @@ def set_frontmatter_field(path: Path, key: str, value: str) -> bool:
     if not path.is_file():
         return False
     text = path.read_text(encoding="utf-8")
-    if not text.startswith("---"):
-        return False
-    parts = text.split("---", 2)
-    if len(parts) < 3:
+    split = _split_frontmatter(text)
+    if split is None:
         return False
-    block_lines = parts[1].splitlines(keepends=True)
+    before, block, after = split
+    block_lines = block.splitlines(keepends=True)
     replaced = False
     for i, line in enumerate(block_lines):
         stripped = line.lstrip()
@@ -1038,7 +934,7 @@ def set_frontmatter_field(path: Path, key: str, value: str) -> bool:
             break
     if not replaced:
         block_lines.append(f"{key}: {value}\n")
-    rebuilt = parts[0] + "---" + "".join(block_lines) + "---" + parts[2]
+    rebuilt = before + "".join(block_lines) + after
     if rebuilt == text:  # already at the target value — idempotent no-op
         return False
     path.write_text(rebuilt, encoding="utf-8")

tests/test_resolve.py

@@ -107,6 +107,37 @@ def test_set_frontmatter_field_replaces_inserts_idempotent(tmp_path):
     assert verify.set_frontmatter_field(bare, "baseline_revision", "abc123") is False
 
 
+def test_set_frontmatter_status_preserves_triple_dash_in_value(tmp_path):
+    """A `---` inside a scalar is not the closing delimiter: status flips and the
+    ---bearing title + body survive (a plain split("---", 2) corrupted this)."""
+    spec = tmp_path / "spec.md"
+    spec.write_text(
+        "---\ntitle: 'restore --- review'\nstatus: in-review\n---\nbody text\n",
+        encoding="utf-8",
+    )
+    assert verify.set_frontmatter_status(spec, "done") is True
+    fm = verify.read_frontmatter(spec)
+    assert fm["status"] == "done"
+    assert fm["title"] == "restore --- review"  # scalar with --- intact
+    assert "body text" in spec.read_text(encoding="utf-8")
+
+
+def test_set_frontmatter_field_preserves_triple_dash_in_value(tmp_path):
+    """Inserting a field appends before the real standalone closing `---`, not
+    inside a scalar that merely contains `---` (which the old split corrupted)."""
+    spec = tmp_path / "spec.md"
+    spec.write_text(
+        "---\ntitle: 'restore --- review'\nstatus: done\n---\nbody text\n",
+        encoding="utf-8",
+    )
+    assert verify.set_frontmatter_field(spec, "baseline_revision", "abc123") is True
+    fm = verify.read_frontmatter(spec)
+    assert fm["baseline_revision"] == "abc123"
+    assert fm["title"] == "restore --- review"
+    assert fm["status"] == "done"
+    assert "body text" in spec.read_text(encoding="utf-8")
+
+
 # ----------------------------------------------------------- build_context
 
 

tests/test_verify.py

@@ -1485,6 +1485,18 @@ def test_read_frontmatter_oserror_still_raises(project, monkeypatch):
         verify.read_frontmatter(p)
 
 
+def test_read_frontmatter_ignores_triple_dash_in_value(project):
+    """A `---` inside a scalar value must NOT be read as the closing delimiter.
+    A plain `split("---", 2)` truncates the block at the first substring match →
+    YAMLError → {}, silently zeroing status; the closing boundary is a standalone
+    `---` line only."""
+    p = project.project / "x.md"
+    p.write_text("---\ntitle: 'restore --- review'\nstatus: done\n---\nbody\n")
+    fm = verify.read_frontmatter(p)
+    assert fm["status"] == "done"
+    assert fm["title"] == "restore --- review"
+
+
 def test_artifact_relpaths_returns_in_repo_folders(project):
     """The orchestrator-owned artifact folders, repo-relative posix."""
     rels = verify.artifact_relpaths(project)