From fbf75413529f3effe22d18c4293404ae1064202c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=A0=E4=B8=BA?= <27188611+Haven2026@users.noreply.github.com> Date: Thu, 23 Jul 2026 17:39:47 +0800 Subject: [PATCH 1/4] fix(deferredwork): keep flat appends visible --- src/bmad_loop/deferredwork.py | 12 +++++++++++- tests/test_deferredwork.py | 18 ++++++++++++++++++ 2 files changed, 29 insertions(+), 1 deletion(-) diff --git a/src/bmad_loop/deferredwork.py b/src/bmad_loop/deferredwork.py index 6c90aaf3..7e80c6b2 100644 --- a/src/bmad_loop/deferredwork.py +++ b/src/bmad_loop/deferredwork.py @@ -18,6 +18,7 @@ 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:[ \t]", re.IGNORECASE | re.MULTILINE) STATUS_RE = re.compile(r"^status:[ \t]*(.*)$", re.MULTILINE) @@ -46,6 +47,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( @@ -144,6 +148,7 @@ def append_entry( origin: str, source_spec: str, reason: str, + location: str = "n/a", status: str = "open", severity: str | None = None, ) -> str | None: @@ -163,7 +168,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 ffa89906..4d29b8b8 100644 --- a/tests/test_deferredwork.py +++ b/tests/test_deferredwork.py @@ -299,6 +299,23 @@ 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_legacy_item_does_not_swallow_masked_canonical_neighbor(): text = ( "## Deferred from: somewhere (2026-06-01)\n\n" @@ -414,6 +431,7 @@ 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 From 9b2efe2fbddc8cfb40d494db7c862d20c85a11ef Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=A0=E4=B8=BA?= <27188611+Haven2026@users.noreply.github.com> Date: Thu, 23 Jul 2026 18:39:33 +0800 Subject: [PATCH 2/4] fix(deferredwork): validate appended locations --- src/bmad_loop/deferredwork.py | 2 ++ tests/test_deferredwork.py | 31 +++++++++++++++++++++++++++++++ 2 files changed, 33 insertions(+) diff --git a/src/bmad_loop/deferredwork.py b/src/bmad_loop/deferredwork.py index 7e80c6b2..03068b32 100644 --- a/src/bmad_loop/deferredwork.py +++ b/src/bmad_loop/deferredwork.py @@ -159,6 +159,8 @@ 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.""" + if "\r" in location or "\n" in location: + raise ValueError("location must be a single line") text = path.read_text(encoding="utf-8") if path.is_file() else "" for entry in parse_ledger(text): if ( diff --git a/tests/test_deferredwork.py b/tests/test_deferredwork.py index 4d29b8b8..089e753b 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, @@ -437,6 +439,35 @@ def test_append_entry_numbers_and_writes(tmp_path): 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 + + +def test_append_entry_rejects_multiline_location_without_writing(tmp_path): + p = tmp_path / "deferred-work.md" + with pytest.raises(ValueError, match="location must be a single line"): + append_entry( + p, + title="follow-up", + origin="code-review", + source_spec="spec-foo.md", + reason="still open", + location="src/foo.py\nstatus: done 2026-07-23", + ) + 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") From 582e5662ba6eaf07128e96236ca7ead4ce5f1dff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=A0=E4=B8=BA?= <27188611+Haven2026@users.noreply.github.com> Date: Thu, 23 Jul 2026 19:18:26 +0800 Subject: [PATCH 3/4] fix(deferredwork): harden canonical serialization --- src/bmad_loop/deferredwork.py | 31 ++++++++++++++++-- tests/test_deferredwork.py | 59 ++++++++++++++++++++++++++++++----- 2 files changed, 80 insertions(+), 10 deletions(-) diff --git a/src/bmad_loop/deferredwork.py b/src/bmad_loop/deferredwork.py index 03068b32..9be0c3bf 100644 --- a/src/bmad_loop/deferredwork.py +++ b/src/bmad_loop/deferredwork.py @@ -18,8 +18,15 @@ 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:[ \t]", re.IGNORECASE | re.MULTILINE) +FLAT_ENTRY_RE = re.compile( + r"^- source_spec:[^\r\n]*\n" + r"[ \t]+summary:[^\r\n]*\n" + r"[ \t]+evidence:[^\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"}) @dataclass(frozen=True) @@ -141,6 +148,11 @@ 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 "\r" in value or "\n" in value: + raise ValueError(f"{name} must be a single line") + + def append_entry( path: Path, *, @@ -159,8 +171,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.""" - if "\r" in location or "\n" in location: - raise ValueError("location must be a single line") + for name, value in ( + ("title", title), + ("origin", origin), + ("source_spec", source_spec), + ("reason", reason), + ("location", location), + ("status", status), + ): + _require_single_line(name, value) + 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") + if CANONICAL_STATUS_RE.fullmatch(status) is None: + raise ValueError("status must be open or done YYYY-MM-DD") text = path.read_text(encoding="utf-8") if path.is_file() else "" for entry in parse_ledger(text): if ( diff --git a/tests/test_deferredwork.py b/tests/test_deferredwork.py index 089e753b..50e0da7f 100644 --- a/tests/test_deferredwork.py +++ b/tests/test_deferredwork.py @@ -318,6 +318,22 @@ def test_flat_append_after_canonical_stays_visible_to_legacy_parser(): 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" @@ -454,16 +470,45 @@ def test_append_entry_preserves_supplied_location(tmp_path): assert "location: src/foo.py:12" in entry.body -def test_append_entry_rejects_multiline_location_without_writing(tmp_path): +@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("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_noncanonical_severity(tmp_path): p = tmp_path / "deferred-work.md" - with pytest.raises(ValueError, match="location must be a single line"): + with pytest.raises(ValueError, match="severity must be critical, high, medium, or low"): append_entry( p, - title="follow-up", - origin="code-review", - source_spec="spec-foo.md", - reason="still open", - location="src/foo.py\nstatus: done 2026-07-23", + title="t", + origin="o", + source_spec="s.md", + reason="r", + severity="urgent", ) assert not p.exists() From 9c6d5c86731790a2b3a01e8d79816ef88a0983e2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=97=A0=E4=B8=BA?= <27188611+Haven2026@users.noreply.github.com> Date: Thu, 23 Jul 2026 20:36:16 +0800 Subject: [PATCH 4/4] fix(deferredwork): guard ledger mutator inputs --- src/bmad_loop/deferredwork.py | 45 ++++++++++-- tests/test_deferredwork.py | 132 ++++++++++++++++++++++++++++++++++ 2 files changed, 171 insertions(+), 6 deletions(-) diff --git a/src/bmad_loop/deferredwork.py b/src/bmad_loop/deferredwork.py index 9be0c3bf..e19a9f24 100644 --- a/src/bmad_loop/deferredwork.py +++ b/src/bmad_loop/deferredwork.py @@ -14,19 +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]*\n" - r"[ \t]+summary:[^\r\n]*\n" - r"[ \t]+evidence:[^\r\n]*(?:\n|$)", + 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) @@ -98,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: @@ -118,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: @@ -149,10 +156,36 @@ def field_line_present(body: str, field: str, value: str) -> bool: def _require_single_line(name: str, value: str) -> None: - if "\r" in value or "\n" in value: + 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, *, @@ -180,12 +213,12 @@ def append_entry( ("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") - if CANONICAL_STATUS_RE.fullmatch(status) is None: - raise ValueError("status must be open or done YYYY-MM-DD") + _require_canonical_status(status) text = path.read_text(encoding="utf-8") if path.is_file() else "" for entry in parse_ledger(text): if ( diff --git a/tests/test_deferredwork.py b/tests/test_deferredwork.py index 50e0da7f..92fa9e0b 100644 --- a/tests/test_deferredwork.py +++ b/tests/test_deferredwork.py @@ -115,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") @@ -138,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, @@ -318,6 +375,33 @@ def test_flat_append_after_canonical_stays_visible_to_legacy_parser(): 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" @@ -491,6 +575,38 @@ def test_append_entry_rejects_multiline_fields_without_writing(tmp_path, field): 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" @@ -499,6 +615,22 @@ def test_append_entry_rejects_noncanonical_status(tmp_path, 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"):