fix(deferredwork): keep flat appends visible after canonical entries

src/bmad_loop/deferredwork.py

@@ -14,11 +14,21 @@
 import hashlib
 import re
 from dataclasses import dataclass
+from datetime import date as calendar_date
 from pathlib import Path
 
 HEADING_RE = re.compile(r"^### (DW-\d+): (.+?)\s*$", re.MULTILINE)
 ANY_HEADING_RE = re.compile(r"^#{1,6} ", re.MULTILINE)
+FLAT_ENTRY_RE = re.compile(
+    r"^- source_spec:[^\r\n]*(?:\r\n|\n)"
+    r"[ \t]+summary:[^\r\n]*(?:\r\n|\n)"
+    r"[ \t]+evidence:[^\r\n]*(?:\r\n|\n|$)",
+    re.IGNORECASE | re.MULTILINE,
+)
 STATUS_RE = re.compile(r"^status:[ \t]*(.*)$", re.MULTILINE)
+CANONICAL_STATUS_RE = re.compile(r"^(?:open|done \d{4}-\d{2}-\d{2})$")
+CANONICAL_SEVERITIES = frozenset({"critical", "high", "medium", "low"})
+LINE_BREAK_RE = re.compile(r"[\n\r\v\f\x1c-\x1e\x85\u2028\u2029]")
 
 
 @dataclass(frozen=True)
@@ -46,6 +56,9 @@ def parse_ledger(text: str) -> list[DWEntry]:
         other = ANY_HEADING_RE.search(text, m.end(), end)
         if other:
             end = other.start()
+        flat = FLAT_ENTRY_RE.search(text, m.end(), end)
+        if flat:
+            end = flat.start()
         body = text[m.start() : end]
         status_m = STATUS_RE.search(body)
         entries.append(
@@ -87,6 +100,8 @@ def mark_done(path: Path, dw_id: str, date: str, note: str) -> bool:
     Returns False (no write) when the entry is missing or already done."""
     if not path.is_file():
         return False
+    _require_iso_date("date", date)
+    _require_single_line("note", note)
     text = path.read_text(encoding="utf-8")
     entry = _find_entry(text, dw_id)
     if entry is None or not entry.open:
@@ -107,6 +122,9 @@ def append_decision(path: Path, dw_id: str, date: str, label: str, detail: str)
     """Record a human decision on an entry without changing its status."""
     if not path.is_file():
         return False
+    _require_iso_date("date", date)
+    _require_single_line("label", label)
+    _require_single_line("detail", detail)
     text = path.read_text(encoding="utf-8")
     entry = _find_entry(text, dw_id)
     if entry is None:
@@ -137,13 +155,45 @@ def field_line_present(body: str, field: str, value: str) -> bool:
     return re.search(rf"(?m)^{re.escape(field)}:[ \t]*`?{v}`?[ \t]*$", body) is not None
 
 
+def _require_single_line(name: str, value: str) -> None:
+    if LINE_BREAK_RE.search(value):
+        raise ValueError(f"{name} must be a single line")
+
+
+def _require_nonempty(name: str, value: str) -> None:
+    if not value.strip():
+        raise ValueError(f"{name} must not be empty")
+
+
+def _require_iso_date(name: str, value: str) -> None:
+    _require_single_line(name, value)
+    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value) is None:
+        raise ValueError(f"{name} must be YYYY-MM-DD")
+    try:
+        calendar_date.fromisoformat(value)
+    except ValueError as error:
+        raise ValueError(f"{name} must be YYYY-MM-DD") from error
+
+
+def _require_canonical_status(status: str) -> None:
+    _require_single_line("status", status)
+    if CANONICAL_STATUS_RE.fullmatch(status) is None:
+        raise ValueError("status must be open or done YYYY-MM-DD")
+    if status.startswith("done "):
+        try:
+            calendar_date.fromisoformat(status.removeprefix("done "))
+        except ValueError as error:
+            raise ValueError("status must be open or done YYYY-MM-DD") from error
+
+
 def append_entry(
     path: Path,
     *,
     title: str,
     origin: str,
     source_spec: str,
     reason: str,
+    location: str = "n/a",
     status: str = "open",
     severity: str | None = None,
 ) -> str | None:
@@ -154,6 +204,21 @@ def append_entry(
     the same `origin:` marker and `source_spec:` — so re-running the same defer
     (e.g. a second sweep of the same story) never duplicates the entry. Creates
     the ledger (and parent dir) if it does not yet exist."""
+    for name, value in (
+        ("title", title),
+        ("origin", origin),
+        ("source_spec", source_spec),
+        ("reason", reason),
+        ("location", location),
+        ("status", status),
+    ):
+        _require_single_line(name, value)
+    _require_nonempty("title", title)
+    if severity is not None:
+        _require_single_line("severity", severity)
+        if severity not in CANONICAL_SEVERITIES:
+            raise ValueError("severity must be critical, high, medium, or low")
+    _require_canonical_status(status)
     text = path.read_text(encoding="utf-8") if path.is_file() else ""
     for entry in parse_ledger(text):
         if (
@@ -163,7 +228,12 @@ def append_entry(
         ):
             return None
     dw_id = f"DW-{next_seq(text)}"
-    lines = [f"### {dw_id}: {title}", f"origin: {origin}", f"source_spec: `{source_spec}`"]
+    lines = [
+        f"### {dw_id}: {title}",
+        f"origin: {origin}",
+        f"location: {location}",
+        f"source_spec: `{source_spec}`",
+    ]
     if severity:
         lines.append(f"severity: {severity}")
     lines.append(f"reason: {reason}")

tests/test_deferredwork.py

@@ -2,6 +2,8 @@
 
 from pathlib import Path
 
+import pytest
+
 from bmad_loop.deferredwork import (
     append_decision,
     append_entry,
@@ -113,6 +115,34 @@ def test_mark_done_missing_entry(tmp_path):
     assert path.read_text(encoding="utf-8") == snapshot
 
 
+@pytest.mark.parametrize(
+    ("date", "note"),
+    [
+        ("2026-06-11\n### DW-99: injected", "fixed"),
+        ("2026-06-11", "fixed\n### DW-99: injected\nstatus: open"),
+        ("2026-06-11", "fixed\u2028### DW-99: injected\u2028status: open"),
+    ],
+)
+def test_mark_done_rejects_line_breaks_without_writing(tmp_path, date, note):
+    path = write_ledger(tmp_path)
+    snapshot = path.read_text(encoding="utf-8")
+
+    with pytest.raises(ValueError, match="must be a single line"):
+        mark_done(path, "DW-1", date, note)
+
+    assert path.read_text(encoding="utf-8") == snapshot
+
+
+def test_mark_done_rejects_impossible_calendar_date_without_writing(tmp_path):
+    path = write_ledger(tmp_path)
+    snapshot = path.read_text(encoding="utf-8")
+
+    with pytest.raises(ValueError, match="date must be YYYY-MM-DD"):
+        mark_done(path, "DW-1", "2026-02-30", "fixed")
+
+    assert path.read_text(encoding="utf-8") == snapshot
+
+
 def test_append_decision(tmp_path):
     path = write_ledger(tmp_path)
     assert append_decision(path, "DW-3", "2026-06-11", "Keep cap", "frozen intent stands")
@@ -136,6 +166,35 @@ def test_append_decision_missing_file(tmp_path):
     assert not mark_done(tmp_path / "nope.md", "DW-1", "2026-06-11", "x")
 
 
+@pytest.mark.parametrize(
+    ("date", "label", "detail"),
+    [
+        ("2026-06-11\n### DW-98: injected", "Keep", "detail"),
+        ("2026-06-11", "Keep\n### DW-98: injected", "detail"),
+        ("2026-06-11", "Keep", "detail\n### DW-98: injected\nstatus: open"),
+        ("2026-06-11", "Keep", "detail\u2028### DW-98: injected\u2028status: open"),
+    ],
+)
+def test_append_decision_rejects_line_breaks_without_writing(tmp_path, date, label, detail):
+    path = write_ledger(tmp_path)
+    snapshot = path.read_text(encoding="utf-8")
+
+    with pytest.raises(ValueError, match="must be a single line"):
+        append_decision(path, "DW-3", date, label, detail)
+
+    assert path.read_text(encoding="utf-8") == snapshot
+
+
+def test_append_decision_rejects_impossible_calendar_date_without_writing(tmp_path):
+    path = write_ledger(tmp_path)
+    snapshot = path.read_text(encoding="utf-8")
+
+    with pytest.raises(ValueError, match="date must be YYYY-MM-DD"):
+        append_decision(path, "DW-3", "2026-02-30", "Keep", "detail")
+
+    assert path.read_text(encoding="utf-8") == snapshot
+
+
 # ------------------------------------------------------------------- legacy
 #
 # Fixtures are condensed verbatim from four real pre-DW project ledgers,
@@ -299,6 +358,66 @@ def test_mixed_ledger_keeps_both_views_separate():
     assert all("DW-" not in e.body for e in entries)
 
 
+def test_flat_append_after_canonical_stays_visible_to_legacy_parser():
+    text = (
+        "# Deferred Work\n\n"
+        "### DW-1: canonical\n\n"
+        "origin: test\nlocation: n/a\nreason: test\nstatus: open\n\n"
+        "- source_spec: `spec-next.md`\n"
+        "  summary: later flat finding\n"
+        "  evidence: must not be swallowed by DW-1\n"
+    )
+
+    (canonical,) = parse_ledger(text)
+    (legacy,) = parse_legacy(text)
+
+    assert "later flat finding" not in canonical.body
+    assert legacy.title == "later flat finding"
+
+
+def test_flat_append_boundary_accepts_crlf():
+    newline = "\r\n"
+    text = newline.join(
+        [
+            "# Deferred Work",
+            "",
+            "### DW-1: canonical",
+            "",
+            "origin: test",
+            "location: n/a",
+            "reason: test",
+            "status: open",
+            "",
+            "- source_spec: `spec-next.md`",
+            "  summary: later flat finding",
+            "  evidence: must not be swallowed by DW-1",
+            "",
+        ]
+    )
+
+    (canonical,) = parse_ledger(text)
+    (legacy,) = parse_legacy(text)
+
+    assert "later flat finding" not in canonical.body
+    assert legacy.title == "later flat finding"
+
+
+def test_source_spec_bullet_without_flat_shape_stays_in_canonical_body():
+    text = (
+        "### DW-1: canonical\n\n"
+        "origin: test\nlocation: n/a\nreason: evidence follows\n"
+        "- source_spec: `quoted-example.md`\n"
+        "  evidence: this is prose evidence, not a flat appender block\n"
+        "status: open\n"
+    )
+
+    (canonical,) = parse_ledger(text)
+
+    assert canonical.open
+    assert "quoted-example.md" in canonical.body
+    assert parse_legacy(text) == []
+
+
 def test_legacy_item_does_not_swallow_masked_canonical_neighbor():
     text = (
         "## Deferred from: somewhere (2026-06-01)\n\n"
@@ -414,11 +533,118 @@ def test_append_entry_numbers_and_writes(tmp_path):
     assert "DW-5" in entries and entries["DW-5"].open
     body = entries["DW-5"].body
     assert "origin: review-budget-followup" in body
+    assert "location: n/a" in body
     assert "source_spec: `spec-foo.md`" in body
     assert "severity: low" in body
     assert "follow-up still recommended for dw-x" in body
 
 
+def test_append_entry_preserves_supplied_location(tmp_path):
+    p = tmp_path / "deferred-work.md"
+    new_id = append_entry(
+        p,
+        title="follow-up",
+        origin="code-review",
+        source_spec="spec-foo.md",
+        reason="still open",
+        location="src/foo.py:12",
+    )
+    assert new_id == "DW-1"
+    (entry,) = parse_ledger(p.read_text())
+    assert "location: src/foo.py:12" in entry.body
+
+
+@pytest.mark.parametrize(
+    "field",
+    ["title", "origin", "source_spec", "reason", "location", "status", "severity"],
+)
+def test_append_entry_rejects_multiline_fields_without_writing(tmp_path, field):
+    p = tmp_path / "deferred-work.md"
+    values = {
+        "title": "follow-up",
+        "origin": "code-review",
+        "source_spec": "spec-foo.md",
+        "reason": "still open",
+        "location": "src/foo.py",
+        "status": "open",
+        "severity": "low",
+    }
+    values[field] += "\nstatus: done 2026-07-23"
+    with pytest.raises(ValueError, match=rf"{field} must be a single line"):
+        append_entry(p, **values)
+    assert not p.exists()
+
+
+@pytest.mark.parametrize(
+    "field",
+    ["title", "origin", "source_spec", "reason", "location", "status", "severity"],
+)
+def test_append_entry_rejects_unicode_line_separator_without_writing(tmp_path, field):
+    p = tmp_path / "deferred-work.md"
+    values = {
+        "title": "follow-up",
+        "origin": "code-review",
+        "source_spec": "spec-foo.md",
+        "reason": "still open",
+        "location": "src/foo.py",
+        "status": "open",
+        "severity": "low",
+    }
+    values[field] += "\u2028status: done 2026-07-23"
+
+    with pytest.raises(ValueError, match=rf"{field} must be a single line"):
+        append_entry(p, **values)
+
+    assert not p.exists()
+
+
+def test_append_entry_rejects_empty_title_without_writing(tmp_path):
+    p = tmp_path / "deferred-work.md"
+
+    with pytest.raises(ValueError, match="title must not be empty"):
+        append_entry(p, title="", origin="o", source_spec="s.md", reason="r")
+
+    assert not p.exists()
+
+
+@pytest.mark.parametrize("status", ["", "open extra", "done", "closed"])
+def test_append_entry_rejects_noncanonical_status(tmp_path, status):
+    p = tmp_path / "deferred-work.md"
+    with pytest.raises(ValueError, match="status must be open or done YYYY-MM-DD"):
+        append_entry(p, title="t", origin="o", source_spec="s.md", reason="r", status=status)
+    assert not p.exists()
+
+
+def test_append_entry_rejects_impossible_done_date_without_writing(tmp_path):
+    p = tmp_path / "deferred-work.md"
+
+    with pytest.raises(ValueError, match="status must be open or done YYYY-MM-DD"):
+        append_entry(
+            p,
+            title="t",
+            origin="o",
+            source_spec="s.md",
+            reason="r",
+            status="done 2026-02-30",
+        )
+
+    assert not p.exists()
+
+
+def test_append_entry_rejects_noncanonical_severity(tmp_path):
+    p = tmp_path / "deferred-work.md"
+    with pytest.raises(ValueError, match="severity must be critical, high, medium, or low"):
+        append_entry(
+            p,
+            title="t",
+            origin="o",
+            source_spec="s.md",
+            reason="r",
+            severity="urgent",
+        )
+    assert not p.exists()
+
+
 def test_append_entry_idempotent_for_open_origin_and_spec(tmp_path):
     p = tmp_path / "deferred-work.md"
     p.write_text("# Deferred Work\n")