diff --git a/src/bmad_loop/deferredwork.py b/src/bmad_loop/deferredwork.py index 6c90aaf..e19a9f2 100644 --- a/src/bmad_loop/deferredwork.py +++ b/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,6 +155,37 @@ 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, *, @@ -144,6 +193,7 @@ def append_entry( 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}") diff --git a/tests/test_deferredwork.py b/tests/test_deferredwork.py index ffa8990..92fa9e0 100644 --- a/tests/test_deferredwork.py +++ b/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")