From 86e09d4bdbac1b997b167891c65707c5971143ba Mon Sep 17 00:00:00 2001 From: t Date: Tue, 25 Aug 2026 21:20:29 -0700 Subject: [PATCH 01/12] fix(deferredwork): serialize ledger mutators on a cross-process lock (#286, #469) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Every deferred-work ledger mutator was an unlocked read-modify-write of the whole file, so two orchestrator processes both read, both edited, and the last atomic write won — losing entries, reverting closures, and letting two appenders mint the same DW- from the same next_seq read. Add runs.lock_path_for (a state-root sidecar keyed on the resolved path) and deferredwork.ledger_lock (public, lazy runs import, thread-local reentrancy guard that raises rather than self-deadlocking on flock's per-fd semantics), and wrap the five leaf mutators: _mark_done_many, mark_open, append_decision, append_entry (closing the next_seq mint race) and archive_closed (one acquisition across both writes). Validation stays above the lock; readers stay lock-free on atomic snapshots. The lock lives out of the repository because the ledger is tracked by design and verify.commit_story/finalize_commit stage with `git add -A`, so a sidecar beside the file would ride into the engine's own commits. --- CHANGELOG.md | 16 + src/bmad_loop/deferredwork.py | 622 ++++++++++++++++++++-------------- src/bmad_loop/runs.py | 34 ++ tests/test_deferredwork.py | 362 +++++++++++++++++++- tests/test_runs.py | 73 ++++ 5 files changed, 849 insertions(+), 258 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dc31b485..4e222391 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -117,6 +117,22 @@ breaking changes may land in a minor release. triage never passes `validate_triage`, and a fresh triage can renumber the option it named — applied by journaled discard rather than by error: the build decision is honored under the always-legal `decision-` fallback name. +- **Deferred-work ledger mutators serialize on a cross-process lock** (#286, #469). Every + mutator was an unlocked read-modify-write of the whole file, so two orchestrator processes — + a second `bmad-loop run`, a run plus a sweep, a run plus the TUI decision modal, a run plus + `sweep --archive` — both read, both edited, and the last atomic write won: entries lost, + closures silently reverted, and two appenders minting the same `DW-` because each read + `next_seq` from the text it had just read. `_mark_done_many`, `mark_open`, `append_decision`, + `append_entry` and `archive_closed` now hold an advisory lock across their whole + read-modify-write, with `archive_closed` covering both of its writes in one acquisition. + Readers stay lock-free — every writer already replaced the file atomically, so a reader sees + one whole version or another. The lock lives at + `/locks/-.lock`, out of the repository rather than beside the + ledger, because the ledger is tracked by design and the engine stages with `git add -A`; it + is keyed on the resolved path, so every spelling of one file contends on one lock. Nested + acquisition raises instead of self-deadlocking, and a lock that cannot be taken fails the + write rather than proceeding unlocked. The dev/review session's own ledger writes are + unchanged and still take no lock. ### Security diff --git a/src/bmad_loop/deferredwork.py b/src/bmad_loop/deferredwork.py index 0d6e8c05..aa798d72 100644 --- a/src/bmad_loop/deferredwork.py +++ b/src/bmad_loop/deferredwork.py @@ -12,21 +12,36 @@ canonical entries. The orchestrator never trusts an LLM to have edited it — status flips and decision records happen here, and gates re-read the file from disk. + +Concurrency (#286/#469): every mutator below is a read->edit->write of the whole +file, so two orchestrator processes — a second `bmad-loop run`, a run plus a +sweep, a run plus the TUI decision modal, a run plus `sweep --archive` — would +otherwise both read, both edit, and let the last atomic write win. Each leaf +mutator therefore runs its whole read->edit->write under :func:`ledger_lock`, a +cross-process mutex on an out-of-repo sidecar. Readers stay lock-free on +purpose: every writer replaces the file atomically, so a reader already sees one +whole version or another, and taking the lock to read would buy nothing while +adding a way to deadlock. Out of scope by #286's own non-goals: the dev/review +LLM session writes this file directly and does NOT take the lock — orchestrator +writes are sequenced against sessions today, so the exposure this closes is +orchestrator-vs-orchestrator. """ from __future__ import annotations import hashlib import re +import threading from bisect import bisect_right -from collections.abc import Sequence +from collections.abc import Iterator, Sequence +from contextlib import contextmanager from dataclasses import dataclass from datetime import date as calendar_date from pathlib import Path from . import sprintstatus from .fences import fenced_spans -from .platform_util import atomic_write_text, neutralize_surrogates +from .platform_util import atomic_write_text, file_lock, neutralize_surrogates HEADING_RE = re.compile(r"^### (DW-\d+): (.+?)\s*$", re.MULTILINE) # Where a canonical entry ENDS, in every shape CommonMark spells an ATX heading: @@ -734,6 +749,57 @@ def _operation_digest(operation_id: str) -> str: return hashlib.sha256(operation_id.encode("utf-8")).hexdigest() +# Per-thread reentrancy guard for :func:`ledger_lock`. `file_lock` is per open +# fd, so a second acquisition from the same process does not merely queue — on +# POSIX `flock` it blocks forever against a lock this very thread holds, with no +# timeout and no traceback. Thread-local rather than a plain module global +# because the state being tracked is "does THIS thread already hold it", and two +# threads legitimately contend through the OS lock. +_LOCK_STATE = threading.local() + + +@contextmanager +def ledger_lock(path: Path) -> Iterator[None]: + """Cross-process mutual exclusion for one ledger (#286/#469). + + Held only around a single read->edit->write of `path` — never across a + subprocess, a coding-CLI session, or an operator pause. That is an acceptance + criterion of #286 rather than a style preference: `file_lock`'s Windows + branch gives up after ~10 s and raises, so a holder that waits on anything + slower converts a contended run into a failed one. It is also why the + engine's rollback/restore windows, which span git spawns, get compare-and-set + semantics instead of a lock around the window. + + Acquired in exactly two strata: the leaf mutators in this module, and the + engine's CAS restores, which do pure in-memory text work under the hold. + Never call a mutator while holding it — every mutator takes this lock itself, + and the nested acquisition would deadlock. + + Nesting raises :class:`RuntimeError` rather than deadlocking. The guard is + deliberately path-agnostic: two *different* ledgers would not self-deadlock + on the OS lock, but nesting is still a lock-ordering hazard, and no caller + has a reason to hold two ledgers at once. The lock file itself lives out of + the repository — see :func:`~bmad_loop.runs.lock_path_for` for why a sidecar + beside the tracked ledger would be committed by the engine's own `git add + -A`. Propagates `OSError` from acquisition and + :class:`~bmad_loop.runs.StateRootError` when no state root can be derived: a + write that could not be serialized must fail loudly, not proceed unlocked. + """ + # Lazy, and it has to stay lazy: `runs` imports `verify`, which imports this + # module, so a top-level import here closes the cycle. + from . import runs + + if getattr(_LOCK_STATE, "held", False): + raise RuntimeError("ledger lock is not reentrant") + lock_path = runs.lock_path_for(path) + _LOCK_STATE.held = True + try: + with file_lock(lock_path): + yield + finally: + _LOCK_STATE.held = False + + def _apply_done( text: str, dw_id: str, @@ -787,23 +853,31 @@ def _mark_done_many( *, operation_id: str | None = None, ) -> list[str]: - """Shared atomic implementation for the public close operations.""" + """Shared atomic implementation for the public close operations. + + The whole read->edit->write runs under the cross-process ledger lock + (#286/#469): concurrent mutators — a second run, a sweep, the TUI decision + modal, ``sweep --archive`` — serialize here rather than trading + last-write-wins. Validation stays ABOVE the lock, so a programmer bug reports + itself without first waiting on another process. + """ _require_iso_date(date) undo_owner = _operation_digest(operation_id) if operation_id is not None else None - if not path.is_file(): - return [] - text = path.read_text(encoding="utf-8") - marked: list[str] = [] - for dw_id in dw_ids: - updated = _apply_done(text, dw_id, date, note, undo_owner=undo_owner) - if updated is None: - continue - text = updated - marked.append(dw_id) - if not marked: - return [] - atomic_write_text(path, text) - return marked + with ledger_lock(path): + if not path.is_file(): + return [] + text = path.read_text(encoding="utf-8") + marked: list[str] = [] + for dw_id in dw_ids: + updated = _apply_done(text, dw_id, date, note, undo_owner=undo_owner) + if updated is None: + continue + text = updated + marked.append(dw_id) + if not marked: + return [] + atomic_write_text(path, text) + return marked def mark_done_many(path: Path, dw_ids: Sequence[str], date: str, note: str) -> list[str]: @@ -874,85 +948,91 @@ def mark_open(path: Path, dw_id: str, note: str, operation_id: str) -> bool: than dropped: the reopened entry is no longer archived, but the body its close moved out still is, and that line is the only thing a later triage has to find it with. + + The whole read->edit->write runs under the cross-process ledger lock + (#286/#469): concurrent mutators — a second run, a sweep, the TUI decision + modal, ``sweep --archive`` — serialize here rather than trading + last-write-wins. """ undo_owner = _operation_digest(operation_id) - if not path.is_file(): - return False - text = path.read_text(encoding="utf-8") - entry = _find_entry(text, dw_id) - if entry is None or entry.open: - return False - if entry.status_span is None: - # parse_ledger deliberately tolerates status-less entries. This primitive - # is later called from _defer, where an AttributeError would crash the run - # instead of completing the deferral. - return False - status_line = entry.body[entry.status_span[0] : entry.status_span[1]] - try: - _require_canonical_status(entry.status) - except ValueError: - # Only a canonical status written by mark_done is eligible for undo. - # Preserve malformed or human-authored statuses for validation/reporting. - return False - res_m = _MARK_DONE_TAIL_RE.match(entry.body, entry.status_span[1]) - if res_m is None: - return False - if res_m.group(1).strip() != _one_line(note).strip() or res_m.group(2) != undo_owner: - return False - if status_line != f"status: done {res_m.group(3)}": - return False - try: - previous_status_line = bytes.fromhex(res_m.group(4)).decode("utf-8") - except (UnicodeDecodeError, ValueError): - return False - if LINE_BREAK_RE.search(previous_status_line): - return False - previous_status_m = STATUS_RE.fullmatch(previous_status_line) - previous_status = previous_status_m.group(1).strip() if previous_status_m else "" - if not previous_status or previous_status.split()[0] != "open": - return False - start = entry.span[0] + entry.status_span[0] - end = entry.span[0] + res_m.end() - # Demote the entry's live `archived:` stamps along with the close they - # describe, rather than deleting them. A stub's stamp says "this body lives - # in the archive file"; once the close is undone the body is here and the - # line is a lie, and leaving it standing is not merely untidy — status + - # undo tail + stamp is the exact `_STUB_BODY_RE` shape, so the next - # reopenable close reconstitutes a stub `archive_closed` skips forever, - # stranding the entry outside every future archive (#711). - # - # Cutting the line outright strands the entry a second way: a stub keeps - # neither `location:` nor `reason:` (`_PRESERVED_FIELD_RE`), so the stamp is - # the reopened entry's ONLY route back to the body, and triage arrives with - # a heading and nothing to triage (#711 review). Renaming the field keeps - # both properties — the value still narrows to the archive block, an id - # owning several once a re-closure is archived too, while the renamed line - # matches neither `_ARCHIVED_FIELD_RE` nor `_STUB_BODY_RE`, so the entry - # reads as live and re-archives normally. Rehydrating the body here - # instead was the alternative and is worse: several blocks per id is by - # design, so a rollback's reopen would have to guess which one, and a wrong - # guess overwrites live content with a stale body. - # - # Cuts are disjoint (an `^archived:` line cannot start inside the status - # line or its adjacent tail) and applied back-to-front so earlier offsets - # stay valid. - cuts = [(start, end, previous_status_line)] - for cut_start, cut_end in _archived_line_spans(entry): - # Everything after the field name — value, spacing and the terminating - # newline — carries over verbatim; the span starts at the anchor, so - # the first colon is the field's own. - stamp = entry.body[cut_start:cut_end].split(":", 1)[1] - cuts.append( - ( - entry.span[0] + cut_start, - entry.span[0] + cut_end, - f"{_ARCHIVED_BODY_FIELD}{stamp}", + with ledger_lock(path): + if not path.is_file(): + return False + text = path.read_text(encoding="utf-8") + entry = _find_entry(text, dw_id) + if entry is None or entry.open: + return False + if entry.status_span is None: + # parse_ledger deliberately tolerates status-less entries. This primitive + # is later called from _defer, where an AttributeError would crash the run + # instead of completing the deferral. + return False + status_line = entry.body[entry.status_span[0] : entry.status_span[1]] + try: + _require_canonical_status(entry.status) + except ValueError: + # Only a canonical status written by mark_done is eligible for undo. + # Preserve malformed or human-authored statuses for validation/reporting. + return False + res_m = _MARK_DONE_TAIL_RE.match(entry.body, entry.status_span[1]) + if res_m is None: + return False + if res_m.group(1).strip() != _one_line(note).strip() or res_m.group(2) != undo_owner: + return False + if status_line != f"status: done {res_m.group(3)}": + return False + try: + previous_status_line = bytes.fromhex(res_m.group(4)).decode("utf-8") + except (UnicodeDecodeError, ValueError): + return False + if LINE_BREAK_RE.search(previous_status_line): + return False + previous_status_m = STATUS_RE.fullmatch(previous_status_line) + previous_status = previous_status_m.group(1).strip() if previous_status_m else "" + if not previous_status or previous_status.split()[0] != "open": + return False + start = entry.span[0] + entry.status_span[0] + end = entry.span[0] + res_m.end() + # Demote the entry's live `archived:` stamps along with the close they + # describe, rather than deleting them. A stub's stamp says "this body lives + # in the archive file"; once the close is undone the body is here and the + # line is a lie, and leaving it standing is not merely untidy — status + + # undo tail + stamp is the exact `_STUB_BODY_RE` shape, so the next + # reopenable close reconstitutes a stub `archive_closed` skips forever, + # stranding the entry outside every future archive (#711). + # + # Cutting the line outright strands the entry a second way: a stub keeps + # neither `location:` nor `reason:` (`_PRESERVED_FIELD_RE`), so the stamp is + # the reopened entry's ONLY route back to the body, and triage arrives with + # a heading and nothing to triage (#711 review). Renaming the field keeps + # both properties — the value still narrows to the archive block, an id + # owning several once a re-closure is archived too, while the renamed line + # matches neither `_ARCHIVED_FIELD_RE` nor `_STUB_BODY_RE`, so the entry + # reads as live and re-archives normally. Rehydrating the body here + # instead was the alternative and is worse: several blocks per id is by + # design, so a rollback's reopen would have to guess which one, and a wrong + # guess overwrites live content with a stale body. + # + # Cuts are disjoint (an `^archived:` line cannot start inside the status + # line or its adjacent tail) and applied back-to-front so earlier offsets + # stay valid. + cuts = [(start, end, previous_status_line)] + for cut_start, cut_end in _archived_line_spans(entry): + # Everything after the field name — value, spacing and the terminating + # newline — carries over verbatim; the span starts at the anchor, so + # the first colon is the field's own. + stamp = entry.body[cut_start:cut_end].split(":", 1)[1] + cuts.append( + ( + entry.span[0] + cut_start, + entry.span[0] + cut_end, + f"{_ARCHIVED_BODY_FIELD}{stamp}", + ) ) - ) - for cut_start, cut_end, replacement in sorted(cuts, reverse=True): - text = text[:cut_start] + replacement + text[cut_end:] - atomic_write_text(path, text) - return True + for cut_start, cut_end, replacement in sorted(cuts, reverse=True): + text = text[:cut_start] + replacement + text[cut_end:] + atomic_write_text(path, text) + return True def append_decision(path: Path, dw_id: str, date: str, label: str, detail: str) -> bool: @@ -971,23 +1051,30 @@ def append_decision(path: Path, dw_id: str, date: str, label: str, detail: str) the reasons documented on :func:`mark_done_many`, plus one this sibling shares with it: a bare ``Path.write_text`` truncates *before* it encodes, so any failure between the two — an unencodable value, ``ENOSPC``, ``EIO`` — leaves a - zero-byte ledger where every entry used to be (#328).""" + zero-byte ledger where every entry used to be (#328). + + The whole read->edit->write runs under the cross-process ledger lock + (#286/#469): concurrent mutators — a second run, a sweep, the TUI decision + modal, ``sweep --archive`` — serialize here rather than trading + last-write-wins. + """ _require_iso_date(date) - if not path.is_file(): - return False - text = path.read_text(encoding="utf-8") - entry = _find_entry(text, dw_id) - if entry is None: - return False - label = _one_line(label) - # Sanitize before the emptiness test, never after: a break-only detail - # collapses to "" and must then drop the separator with it, or the entry - # carries a dangling `— ` promising a detail that is not there. - detail = _one_line(detail) - detail_part = f" — {detail}" if detail else "" - text = _insert_after_status(text, entry, f"decision: {date} {label}{detail_part}") - atomic_write_text(path, text) - return True + with ledger_lock(path): + if not path.is_file(): + return False + text = path.read_text(encoding="utf-8") + entry = _find_entry(text, dw_id) + if entry is None: + return False + label = _one_line(label) + # Sanitize before the emptiness test, never after: a break-only detail + # collapses to "" and must then drop the separator with it, or the entry + # carries a dangling `— ` promising a detail that is not there. + detail = _one_line(detail) + detail_part = f" — {detail}" if detail else "" + text = _insert_after_status(text, entry, f"decision: {date} {label}{detail_part}") + atomic_write_text(path, text) + return True DW_ID_RE = re.compile(r"\bDW-(\d+)\b") @@ -1040,7 +1127,15 @@ def append_entry( the reasons documented on :func:`mark_done_many`, plus one this sibling shares with it: a bare ``Path.write_text`` truncates *before* it encodes, so any failure between the two — an unencodable value, ``ENOSPC``, ``EIO`` — leaves a - zero-byte ledger where every entry used to be (#328).""" + zero-byte ledger where every entry used to be (#328). + + The whole read->edit->write runs under the cross-process ledger lock + (#286/#469): concurrent mutators — a second run, a sweep, the TUI decision + modal, ``sweep --archive`` — serialize here rather than trading + last-write-wins. That hold spans the ``next_seq`` mint as + well as the idempotence scan, which is what stops two concurrent appenders + reading the same highest id and both minting it (#469). + """ _require_canonical_status(status) # The whitelist is derived from the legacy parser's alias table (defined # below; resolved at call time) so what this writer emits and what @@ -1053,53 +1148,57 @@ def append_entry( source_spec = _one_line(source_spec) reason = _one_line(reason) location = _one_line(location) - text = path.read_text(encoding="utf-8") if path.is_file() else "" - for entry in parse_ledger(text): - if ( - entry.open - and field_line_present(entry.body, "origin", origin) - and field_line_present(entry.body, "source_spec", source_spec) - ): - return None - dw_id = f"DW-{next_seq(text)}" - if given_title and not title.strip(): - # A break-only title sanitizes to nothing, and `### DW-: ` is a - # heading `HEADING_RE`'s `(.+?)` does not match: the caller is handed an - # id no reader can find while `next_seq` has already burned it. - # - # Tested with `.strip()`, not `not title`: a title of `" "` carries no - # break at all, so `_one_line` returns it unchanged by the byte-identity - # fast path and it stays truthy. It parses, but renders blank in - # `status`, `--json` and the TUI — the unidentifiable half of the same - # problem, reached without ever touching the sanitizer. - # - # Scoped to a title that *had* content: an already-empty one keeps its - # long-standing behavior, and the invariant is about non-empty values. - title = f"(untitled {dw_id})" - 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}") - lines.append(f"status: {status}") - block = "\n".join(lines) + "\n" - # exactly one blank line between the previous content and the new entry - if text == "" or text.endswith("\n\n"): - sep = "" - elif text.endswith("\n"): - sep = "\n" - else: - sep = "\n\n" - path.parent.mkdir(parents=True, exist_ok=True) - atomic_write_text(path, text + sep + block) - return dw_id + with ledger_lock(path): + text = path.read_text(encoding="utf-8") if path.is_file() else "" + for entry in parse_ledger(text): + if ( + entry.open + and field_line_present(entry.body, "origin", origin) + and field_line_present(entry.body, "source_spec", source_spec) + ): + return None + dw_id = f"DW-{next_seq(text)}" + if given_title and not title.strip(): + # A break-only title sanitizes to nothing, and `### DW-: ` is a + # heading `HEADING_RE`'s `(.+?)` does not match: the caller is handed an + # id no reader can find while `next_seq` has already burned it. + # + # Tested with `.strip()`, not `not title`: a title of `" "` carries no + # break at all, so `_one_line` returns it unchanged by the byte-identity + # fast path and it stays truthy. It parses, but renders blank in + # `status`, `--json` and the TUI — the unidentifiable half of the same + # problem, reached without ever touching the sanitizer. + # + # Scoped to a title that *had* content: an already-empty one keeps its + # long-standing behavior, and the invariant is about non-empty values. + title = f"(untitled {dw_id})" + 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}") + lines.append(f"status: {status}") + block = "\n".join(lines) + "\n" + # exactly one blank line between the previous content and the new entry + if text == "" or text.endswith("\n\n"): + sep = "" + elif text.endswith("\n"): + sep = "\n" + else: + sep = "\n\n" + path.parent.mkdir(parents=True, exist_ok=True) + atomic_write_text(path, text + sep + block) + return dw_id ARCHIVE_REL = "deferred-work-archive.md" +# The archive sibling is never locked in its own right: :func:`archive_closed` +# is the only writer, and it holds the LEDGER's :func:`ledger_lock` across both +# writes (#286/#469). Any future writer of this file must take that same lock. # A stub left by a prior archive_closed run carries this field. The next run # reads it to skip entries whose body has already been moved — without it, # every run would re-archive the stub (a heading + status line) and the @@ -1301,115 +1400,124 @@ def archive_closed( skipped by their exact stub shape. A stub's ``archived:`` date names the archive block holding its body, so an entry recovered from a crashed run is stamped with the date already on that block rather than with this run's. + + The whole read->edit->write runs under the cross-process ledger lock + (#286/#469): concurrent mutators — a second run, a sweep, the TUI decision + modal, ``sweep --archive`` — serialize here rather than trading + last-write-wins. ONE acquisition spans BOTH writes — the + archive sibling has no lock of its own precisely because it is only ever + written under its ledger's lock — and ``dry_run`` runs inside the hold too, + so there is one code path rather than a locked and an unlocked one. """ if before is not None: _require_iso_date(before) if archive_date is not None: _require_iso_date(archive_date) - if not path.is_file(): - return [] - text = path.read_text(encoding="utf-8") - to_archive: list[tuple[DWEntry, str]] = [] - for entry in parse_ledger(text): - close_date = _close_date(entry) - if close_date is None: - continue # not done, or done without a date - if before is not None and close_date >= before: - continue # closed on or after the cutoff - if _is_stub(entry): - continue # stub from a prior archive_closed run - to_archive.append((entry, close_date)) - if not to_archive: - return [] - archived_ids = [e.id for e, _ in to_archive] - if dry_run: - return archived_ids - stamp = archive_date or calendar_date.today().isoformat() - archive_path = path.parent / ARCHIVE_REL - existing = archive_path.read_text(encoding="utf-8") if archive_path.is_file() else "" - # Append an `archived:` line after each entry's status line. The status - # span is body-relative, so the insertion works within the body slice — - # same offset math as `_insert_after_status`, applied to the body. - # - # Crash recovery: the archive is written BEFORE the ledger (see below), so - # a crash between the two writes leaves the ledger with full entries whose - # bodies are already in the archive. A retry must still stub those ledger - # entries (completing the interrupted operation) but must NOT append their - # bodies again — an append-only archive accumulating duplicates. Entries - # whose parsed archive twin carries a live (non-fenced) ``archived:`` - # field are therefore skipped here and only replaced with stubs below. - # - # The twin must match in BODY, not merely in id and close date. A DW id is - # reusable across closures (`mark_open` reopens, a re-close follows) and a - # closed entry still accepts writes (`append_decision` does not read - # status), so id + date names a *closure slot*, not its content: reopened - # and re-closed the same day with a new resolution, or annotated with a - # decision after its body was archived, the ledger entry and its twin - # differ. Skipping on the slot alone stubbed that entry over its own - # content while reporting the id as archived — the body reached neither - # file (#711). A body that differs is appended instead; the archive holds - # several blocks per id by design, and over-archiving is recoverable where - # a silent drop is not. - archive_blocks: list[str] = [] - already_archived = { - e.id: ((_close_date(e), _body_without_archived(e)), _archived_stamp(e)) - for e in parse_ledger(existing) - if _is_archived(e) - } # fence-aware: a quoted example in the archive is not a real body - # A recovered entry's stub is stamped with the date already on its archived - # body, not with this run's. The two diverge whenever the retry lands on a - # later day than the crashed run, and the stamp is not decoration: it is - # what picks one of an id's several archive blocks — for a reader following - # the stub, and for the `archived-body:` pointer `mark_open` demotes that - # stamp into, which is a reopened entry's only route back to its body - # (#711 review). A stub naming a date no block carries resolves to nothing. - recovered_stamps: dict[str, str] = {} - for entry, close_date in to_archive: - twin = already_archived.get(entry.id) - if twin is not None and twin[0] == (close_date, _body_without_archived(entry)): - # this closure's body is already archived (crashed prior run) - if twin[1] is not None: - recovered_stamps[entry.id] = twin[1] - continue - body = entry.body - assert entry.status_span is not None # done with a date implies a status line - pos = entry.status_span[1] - body = body[:pos] + f"\narchived: {stamp}" + body[pos:] - archive_blocks.append(body) - # Appended, never prepended: for one id the file's order is closure order, - # which is the documented tie-break when two closures were archived on the - # same day and so carry the same stamp (#711 review). - if archive_blocks: - if existing == "" or existing.endswith("\n\n"): - sep = "" - elif existing.endswith("\n"): - sep = "\n" + with ledger_lock(path): + if not path.is_file(): + return [] + text = path.read_text(encoding="utf-8") + to_archive: list[tuple[DWEntry, str]] = [] + for entry in parse_ledger(text): + close_date = _close_date(entry) + if close_date is None: + continue # not done, or done without a date + if before is not None and close_date >= before: + continue # closed on or after the cutoff + if _is_stub(entry): + continue # stub from a prior archive_closed run + to_archive.append((entry, close_date)) + if not to_archive: + return [] + archived_ids = [e.id for e, _ in to_archive] + if dry_run: + return archived_ids + stamp = archive_date or calendar_date.today().isoformat() + archive_path = path.parent / ARCHIVE_REL + existing = archive_path.read_text(encoding="utf-8") if archive_path.is_file() else "" + # Append an `archived:` line after each entry's status line. The status + # span is body-relative, so the insertion works within the body slice — + # same offset math as `_insert_after_status`, applied to the body. + # + # Crash recovery: the archive is written BEFORE the ledger (see below), so + # a crash between the two writes leaves the ledger with full entries whose + # bodies are already in the archive. A retry must still stub those ledger + # entries (completing the interrupted operation) but must NOT append their + # bodies again — an append-only archive accumulating duplicates. Entries + # whose parsed archive twin carries a live (non-fenced) ``archived:`` + # field are therefore skipped here and only replaced with stubs below. + # + # The twin must match in BODY, not merely in id and close date. A DW id is + # reusable across closures (`mark_open` reopens, a re-close follows) and a + # closed entry still accepts writes (`append_decision` does not read + # status), so id + date names a *closure slot*, not its content: reopened + # and re-closed the same day with a new resolution, or annotated with a + # decision after its body was archived, the ledger entry and its twin + # differ. Skipping on the slot alone stubbed that entry over its own + # content while reporting the id as archived — the body reached neither + # file (#711). A body that differs is appended instead; the archive holds + # several blocks per id by design, and over-archiving is recoverable where + # a silent drop is not. + archive_blocks: list[str] = [] + already_archived = { + e.id: ((_close_date(e), _body_without_archived(e)), _archived_stamp(e)) + for e in parse_ledger(existing) + if _is_archived(e) + } # fence-aware: a quoted example in the archive is not a real body + # A recovered entry's stub is stamped with the date already on its archived + # body, not with this run's. The two diverge whenever the retry lands on a + # later day than the crashed run, and the stamp is not decoration: it is + # what picks one of an id's several archive blocks — for a reader following + # the stub, and for the `archived-body:` pointer `mark_open` demotes that + # stamp into, which is a reopened entry's only route back to its body + # (#711 review). A stub naming a date no block carries resolves to nothing. + recovered_stamps: dict[str, str] = {} + for entry, close_date in to_archive: + twin = already_archived.get(entry.id) + if twin is not None and twin[0] == (close_date, _body_without_archived(entry)): + # this closure's body is already archived (crashed prior run) + if twin[1] is not None: + recovered_stamps[entry.id] = twin[1] + continue + body = entry.body + assert entry.status_span is not None # done with a date implies a status line + pos = entry.status_span[1] + body = body[:pos] + f"\narchived: {stamp}" + body[pos:] + archive_blocks.append(body) + # Appended, never prepended: for one id the file's order is closure order, + # which is the documented tie-break when two closures were archived on the + # same day and so carry the same stamp (#711 review). + if archive_blocks: + if existing == "" or existing.endswith("\n\n"): + sep = "" + elif existing.endswith("\n"): + sep = "\n" + else: + sep = "\n\n" + archive_content = existing + sep + "".join(archive_blocks) else: - sep = "\n\n" - archive_content = existing + sep + "".join(archive_blocks) - else: - archive_content = existing # pure crash-recovery pass: only stub the ledger - # Replace each archived entry's span with a stub, working backwards so - # earlier spans are unaffected by later replacements — the same - # text-surgery pattern as `_apply_done`, applied to multiple entries. - for entry, close_date in reversed(to_archive): - preserved = "".join(f"{line}\n" for line in _preserved_stub_lines(entry)) - stub = ( - f"### {entry.id}: {entry.title}\n\n" - f"status: done {close_date}\n" - f"{preserved}" - f"archived: {recovered_stamps.get(entry.id, stamp)}\n\n" - ) - start, end = entry.span - text = text[:start] + stub + text[end:] - # Write the archive BEFORE the ledger: a crash between writes leaves the - # archive with extra content (harmless — the archive is append-only) and - # the ledger unchanged (safe — the bodies are still in the live file). - # Writing the ledger first would leave stubs in the ledger with no bodies - # in the archive — content lost. - atomic_write_text(archive_path, archive_content) - atomic_write_text(path, text) - return archived_ids + archive_content = existing # pure crash-recovery pass: only stub the ledger + # Replace each archived entry's span with a stub, working backwards so + # earlier spans are unaffected by later replacements — the same + # text-surgery pattern as `_apply_done`, applied to multiple entries. + for entry, close_date in reversed(to_archive): + preserved = "".join(f"{line}\n" for line in _preserved_stub_lines(entry)) + stub = ( + f"### {entry.id}: {entry.title}\n\n" + f"status: done {close_date}\n" + f"{preserved}" + f"archived: {recovered_stamps.get(entry.id, stamp)}\n\n" + ) + start, end = entry.span + text = text[:start] + stub + text[end:] + # Write the archive BEFORE the ledger: a crash between writes leaves the + # archive with extra content (harmless — the archive is append-only) and + # the ledger unchanged (safe — the bodies are still in the live file). + # Writing the ledger first would leave stubs in the ledger with no bodies + # in the archive — content lost. + atomic_write_text(archive_path, archive_content) + atomic_write_text(path, text) + return archived_ids # ------------------------------------------------------------------- legacy diff --git a/src/bmad_loop/runs.py b/src/bmad_loop/runs.py index 2f6c3943..a6d2fa5f 100644 --- a/src/bmad_loop/runs.py +++ b/src/bmad_loop/runs.py @@ -926,6 +926,40 @@ def accepted_tags(project: Path) -> frozenset[str]: return frozenset({project_tag(project), str(project.resolve())}) +def lock_path_for(data_path: Path) -> Path: + """The advisory-lock sidecar for a mutable data file: + ``/locks/-.lock``. + + Out of the repository, deliberately, and NOT the ``.lock`` sibling the + obvious reading of #286 asks for. The deferred-work ledger is a *tracked* + file by design, and both :func:`verify.commit_story` and + :func:`verify.finalize_commit` stage with ``git add -A``: a lock beside it + would be swept into the engine's own commits, and the git-add shield that + would otherwise hide it covers linked worktrees only. Under the state root + the sidecar is never git-visible at all, so no exclusion machinery has to be + kept correct for it. + + Keyed on the **resolved** path so the identity of the lock is the identity of + the file rather than of the spelling used to reach it: a symlinked and a + direct path to one ledger rendezvous on one lock (without which the two + spellings would exclude nobody), two worktrees' in-tree ledgers are different + files and correctly get independent locks, and several projects pointed at a + shared external artifact dir land on one lock, which is where the real + contention is. The basename is appended for debuggability only — a human + reading ``ls`` of the locks dir should see which file a sidecar guards — and + carries no meaning for exclusion, which rides the digest. + + Pure: no directory is created here, because + :func:`~bmad_loop.platform_util.file_lock` mkdirs the parent when it opens + the lock. May raise :class:`StateRootError` when the environment names no + usable state root (see :func:`state_root`); the caller fails rather than + silently locking somewhere else. + """ + resolved = data_path.resolve() + digest = hashlib.sha256(os.fsencode(str(resolved))).hexdigest()[:16] + return state_root() / "locks" / f"{digest}-{resolved.name}.lock" + + def mux_sessions() -> list[str]: """All live session names, or [] when the multiplexer is missing, no server is running, or the query fails.""" diff --git a/tests/test_deferredwork.py b/tests/test_deferredwork.py index fc578f5e..549ceb36 100644 --- a/tests/test_deferredwork.py +++ b/tests/test_deferredwork.py @@ -1,10 +1,14 @@ """Ledger parsing and editing: deferredwork.py.""" +import contextlib +import subprocess +import sys +import time from pathlib import Path import pytest -from bmad_loop import deferredwork, fences +from bmad_loop import deferredwork, fences, platform_util, runs from bmad_loop.deferredwork import ( _ISO_DATE_RE, ARCHIVE_REL, @@ -3167,3 +3171,359 @@ def test_archive_fenced_archived_line_in_twin_does_not_suppress(tmp_path): # ...and the body left the ledger for the archive rather than being dropped. stub = {e.id: e for e in parse_ledger(path.read_text(encoding="utf-8"))}["DW-2"] assert "reason: pre-existing." not in stub.body + + +# ----------------------------------- cross-process ledger lock (#286, #469) +# +# Every mutator here is a read->edit->write of the whole ledger, so two +# orchestrator processes — a second run, a sweep, the TUI decision modal, +# `sweep --archive` — would otherwise both read, both edit, and let the last +# atomic write win. The section grades four separable claims: that each leaf +# mutator holds `ledger_lock` across its whole critical section, that the hold +# really excludes, that a nested acquisition raises rather than self-deadlocking, +# and that a failed acquisition raises without writing. +# +# Exclusion is probed with `blocking=False` only. That is not an optimization: +# `file_lock` is per open fd, so a blocking probe from this process would wait +# forever on POSIX `flock` against a lock this very thread holds, and ~10s on +# Windows before raising. The suite runs under xdist, so neither is acceptable. + + +def _lock_is_held(path: Path) -> bool: + """True when the ledger's sidecar lock cannot be taken right now.""" + try: + with platform_util.file_lock(runs.lock_path_for(path), blocking=False): + return False + except OSError: + return True + + +@contextlib.contextmanager +def _unavailable_lock(path, **kwargs): + """A `file_lock` that cannot be acquired. + + `OSError(11, "Resource deadlock avoided")` is the shape `msvcrt.locking` + raises when its ~10 s blocking retry runs out — a routine outcome on the + Windows legs rather than a contrived one. The dead `yield` after the raise + keeps this a generator function, which `contextlib.contextmanager` requires. + """ + raise OSError(11, "Resource deadlock avoided") + yield # pragma: no cover — unreachable + + +LOCKED_MUTATORS = { + "append_decision": lambda p: append_decision(p, "DW-1", "2026-06-11", "keep", "later"), + "append_entry": lambda p: append_entry( + p, title="new", origin="probe", source_spec="spec-probe.md", reason="raced" + ), + "archive_closed": lambda p: archive_closed(p, archive_date="2026-08-24"), + "mark_done": lambda p: mark_done(p, "DW-1", "2026-06-11", "fixed"), + "mark_done_many": lambda p: mark_done_many(p, ["DW-1"], "2026-06-11", "fixed"), + "mark_done_many_reopenable": lambda p: mark_done_many_reopenable( + p, ["DW-1"], "2026-06-11", "fixed", OPERATION_ID + ), + "mark_open": lambda p: mark_open(p, "DW-1", "by dw-a", OPERATION_ID), +} + + +def _seed_for(tmp_path: Path, name: str) -> Path: + """The ledger `name`'s call needs, written before any lock spy is installed.""" + path = write_ledger(tmp_path) + if name == "mark_open": + close_reopenable(path, "DW-1", "by dw-a") + return path + + +@pytest.mark.parametrize("name", sorted(LOCKED_MUTATORS)) +def test_every_mutator_holds_the_ledger_lock(tmp_path, monkeypatch, name): + """Each leaf mutator takes the lock exactly once, and the hold really excludes. + + Two claims in one assertion, and both are needed. That the spy fired says the + mutator routes through `ledger_lock` at all; that it fired ONCE says the whole + read->edit->write sits inside a single acquisition rather than a per-step + hold that another writer can slip between. The probe inside the critical + section says the acquisition is a real OS lock and not a no-op — a + `ledger_lock` that yielded without taking anything would satisfy the call + count and exclude nobody. + + Ablation: delete this mutator's `with ledger_lock(path):` and dedent its + body — the spy never fires, `probed` stays empty, and the row reds.""" + path = _seed_for(tmp_path, name) + real_lock = deferredwork.ledger_lock + probed = [] + + @contextlib.contextmanager + def spy_lock(p): + with real_lock(p): + probed.append(_lock_is_held(p)) + yield + + monkeypatch.setattr(deferredwork, "ledger_lock", spy_lock) + + LOCKED_MUTATORS[name](path) + + assert probed == [True] + + +def test_ledger_lock_is_not_reentrant(tmp_path, monkeypatch): + """A nested acquisition raises rather than deadlocking, and raises BEFORE it + reaches the OS lock. + + `file_lock` is per open fd: on POSIX a second `flock(LOCK_EX)` from this same + thread blocks against the lock the thread already holds, with no timeout and + no traceback — the run simply stops. The guard converts that silent wedge + into a loud error at the call site that introduced the nesting. + + The `file_lock` counter is what makes the test deterministic. Asserting only + the `RuntimeError` would leave a version of the guard that raises *after* + attempting the acquire indistinguishable from one that raises before it, and + the former hangs. Counting proves the nested entry never reached the kernel, + so this test never risks the deadlock it is about. + + Ablation is deliberately NOT run here: dropping the depth guard makes the + nested entry block forever rather than fail, which hangs the suite instead + of reddening one row. The guard's absence is graded by inspection. + + The tail grades the release: the guard is per-thread state, so a hold that + is not cleared on exit would refuse every later mutation in this thread.""" + path = write_ledger(tmp_path) + real_file_lock = deferredwork.file_lock + acquired = [] + + @contextlib.contextmanager + def counting(lock_path, **kwargs): + acquired.append(lock_path) + with real_file_lock(lock_path, **kwargs): + yield + + monkeypatch.setattr(deferredwork, "file_lock", counting) + + with deferredwork.ledger_lock(path): + with pytest.raises(RuntimeError, match="not reentrant"): + with deferredwork.ledger_lock(path): + pass # pragma: no cover — the guard raises on entry + assert len(acquired) == 1 # the nested entry never reached the OS lock + + with deferredwork.ledger_lock(path): # released cleanly, so this is fine + pass + assert len(acquired) == 2 + + +def test_a_failed_acquisition_does_not_leak_the_reentrancy_guard(tmp_path, monkeypatch): + """An acquisition that raises must still leave this thread unmarked. + + The guard is set before the acquire (it has to be — the acquire is what would + deadlock), so clearing it anywhere but a `finally` strands the thread: every + later mutation in this process raises `RuntimeError` and the run dies of a + transient lock failure it should merely have reported. + + Ablation: move `_LOCK_STATE.held = False` out of `ledger_lock`'s `finally` + onto the success path — the second `mark_done` raises `RuntimeError` instead + of closing DW-1.""" + path = write_ledger(tmp_path) + with monkeypatch.context() as m: + m.setattr(deferredwork, "file_lock", _unavailable_lock) + with pytest.raises(OSError, match="Resource deadlock avoided"): + mark_done(path, "DW-1", "2026-06-11", "fixed") + + assert mark_done(path, "DW-1", "2026-06-11", "fixed") + + +def test_scripted_interleave_loses_no_update(tmp_path, monkeypatch): + """The #286 lost-update scenario, made deterministic: a rival writer commits + in full between writer A's call and A's acquisition, and A must still see it. + + Writer A closes DW-1; writer B appends a new entry. B is run to completion — + acquire, read, write, release — immediately BEFORE A delegates to the real + lock, which is the worst legal interleaving the lock permits. A therefore has + to read the ledger B just wrote, not one it snapshotted earlier, or A's write + reverts B's append. + + Ablation: hoist `_mark_done_many`'s `path.read_text` above its + `with ledger_lock(path):` — A's read then happens before the spy fires, A + writes its stale snapshot, and DW-4 is gone from the final ledger.""" + path = write_ledger(tmp_path) + real_lock = deferredwork.ledger_lock + rival_ran = [] + + @contextlib.contextmanager + def rival_first(p): + if not rival_ran: + rival_ran.append(True) # once: B's own append re-enters this spy + assert ( + append_entry( + p, + title="rival append", + origin="rival-origin", + source_spec="spec-rival.md", + reason="raced with a close", + ) + == "DW-4" + ) + with real_lock(p): + yield + + monkeypatch.setattr(deferredwork, "ledger_lock", rival_first) + + assert mark_done(path, "DW-1", "2026-06-11", "closed by A") + + entries = {e.id: e for e in parse_ledger(path.read_text(encoding="utf-8"))} + assert entries["DW-4"].open # B's append survived A's write + assert "origin: rival-origin" in entries["DW-4"].body + assert entries["DW-1"].status == "done 2026-06-11" # ...and A's close landed + assert "resolution: closed by A" in entries["DW-1"].body + assert len(entries) == 4 # DW-1..DW-3 plus B's, every id distinct + + +def test_archive_closed_writes_both_files_inside_one_acquisition(tmp_path, monkeypatch): + """The archive and the trimmed ledger are written under ONE hold. + + The pair is a transaction: the archive is written first so a crash between + the writes leaves bodies duplicated (harmless, the archive is append-only) + rather than lost. Release the lock between them and a rival mutator lands in + the gap and writes the untrimmed ledger back, resurrecting entries whose + bodies have already moved to the archive — a duplicate no later run cleans + up. It is also why the archive sibling has no lock of its own: it is only + ever written under its ledger's. + + Ablation: hoist either `atomic_write_text` out of the `with` — the event + order changes and the assertion reds.""" + path = write_ledger(tmp_path) + archive = path.parent / ARCHIVE_REL + real_lock, real_write = deferredwork.ledger_lock, deferredwork.atomic_write_text + events = [] + + @contextlib.contextmanager + def spy_lock(p): + events.append("lock-enter") + with real_lock(p): + yield + events.append("lock-exit") + + def spy_write(p, text): + events.append("write-archive" if p == archive else "write-ledger") + return real_write(p, text) + + monkeypatch.setattr(deferredwork, "ledger_lock", spy_lock) + monkeypatch.setattr(deferredwork, "atomic_write_text", spy_write) + + assert archive_closed(path, archive_date="2026-08-24") == ["DW-2"] + + assert events == ["lock-enter", "write-archive", "write-ledger", "lock-exit"] + + +@pytest.mark.parametrize("name", sorted(LOCKED_MUTATORS)) +def test_lock_acquisition_failure_raises_and_writes_nothing(tmp_path, monkeypatch, name): + """A lock that cannot be taken fails the write; it never proceeds unlocked. + + This pins the repo's "repair writes must raise" doctrine at the new seam. + Degrading to an unlocked write would be the worst of both worlds: the caller + is told the mutation succeeded while the exact interleaving the lock exists + to prevent is back, and only under contention — the case no test would catch. + + Ablation: swallow the acquisition `OSError` inside `ledger_lock` and let the + body run anyway — `pytest.raises` fails on every row.""" + path = _seed_for(tmp_path, name) + archive = path.parent / ARCHIVE_REL + before = path.read_text(encoding="utf-8") + archive_before = archive.read_text(encoding="utf-8") if archive.is_file() else None + + monkeypatch.setattr(deferredwork, "file_lock", _unavailable_lock) + + with pytest.raises(OSError, match="Resource deadlock avoided"): + LOCKED_MUTATORS[name](path) + + assert path.read_text(encoding="utf-8") == before + assert (archive.read_text(encoding="utf-8") if archive.is_file() else None) == archive_before + + +# The child of the two-process acceptance test. Appends 8 entries with distinct +# origins (so the idempotence scan never dedupes one away) through the real +# `append_entry`, after a file rendezvous with the parent. It reports the ids it +# minted so the parent can grade the mint, not merely the entry count. +CONCURRENT_APPENDER = ( + "import pathlib, sys, time\n" + "from bmad_loop.deferredwork import append_entry\n" + "ledger, ready, go, done = (pathlib.Path(a) for a in sys.argv[1:5])\n" + "tag = sys.argv[5]\n" + "ready.write_text('ready')\n" + "deadline = time.monotonic() + 60\n" + "while time.monotonic() < deadline and not go.exists():\n" + " time.sleep(0.01)\n" + "if not go.exists():\n" + " sys.exit(3)\n" # never released — what a broken rendezvous looks like + "ids = []\n" + "for n in range(8):\n" + " ids.append(append_entry(ledger, title=tag + '-' + str(n),\n" + " origin=tag + '-origin-' + str(n),\n" + " source_spec='spec-' + tag + '-' + str(n) + '.md',\n" + " reason='raced'))\n" + "done.write_text('\\n'.join(str(i) for i in ids))\n" +) + + +def test_two_processes_append_concurrently_produce_distinct_ids(tmp_path): + """#286's first acceptance criterion, end to end: two PROCESSES appending at + once produce every entry, each with its own id, and lose none. + + This is the integration proof that the lock crosses a process boundary — the + one thing no in-process spy can show, and the only test here that exercises + the `msvcrt` branch on the Windows CI legs rather than `flock`. The children + inherit the environment, so the autouse `_isolate_state_root` fixture's + `BMAD_LOOP_STATE_DIR` reaches them and all three processes resolve the same + sidecar. + + Deliberately NOT this test's job to grade the lock's absence: without it the + outcome is stochastic — a lost update or a duplicated `next_seq` mint needs + the two read->write windows to actually overlap — so an ablation here reds + only sometimes. The deterministic coverage is + `test_every_mutator_holds_the_ledger_lock` and + `test_scripted_interleave_loses_no_update` above. + + No parent-held lock: the parent must not be a third contender, or the + children's blocking acquires would sit on the Windows ~10 s ceiling. Waits + are bounded polls on a monotonic deadline, never a bare sleep.""" + path = write_ledger(tmp_path, "# Deferred Work\n") + go = tmp_path / "go" + procs, readies, dones = [], [], [] + for n in (1, 2): + ready, done = tmp_path / f"ready-{n}", tmp_path / f"done-{n}" + procs.append( + subprocess.Popen( + [ + sys.executable, + "-c", + CONCURRENT_APPENDER, + str(path), + str(ready), + str(go), + str(done), + f"w{n}", + ] + ) + ) + readies.append(ready) + dones.append(done) + try: + deadline = time.monotonic() + 60 + while time.monotonic() < deadline and not all(r.exists() for r in readies): + time.sleep(0.02) + assert all(r.exists() for r in readies), "a child never reached the rendezvous" + + go.write_text("go") # release both at once + for proc in procs: + proc.communicate(timeout=120) + assert proc.returncode == 0 + finally: + for proc in procs: + if proc.poll() is None: + proc.kill() + proc.wait(timeout=10) + + entries = parse_ledger(path.read_text(encoding="utf-8")) + assert len(entries) == 16 # nothing lost to a last-write-wins overwrite + assert len({e.id for e in entries}) == 16 # ...and no id minted twice + + reported = [line for d in dones for line in d.read_text(encoding="utf-8").splitlines()] + assert "None" not in reported # no append was silently deduped away + assert sorted(reported) == sorted(e.id for e in entries) diff --git a/tests/test_runs.py b/tests/test_runs.py index 81145624..1f72b9dc 100644 --- a/tests/test_runs.py +++ b/tests/test_runs.py @@ -1845,6 +1845,79 @@ def test_state_dir_for_raises_when_the_project_cannot_be_canonicalized(tmp_path, runs.state_dir_for(project, "r1") +# ------------------------------------------------- lock_path_for (#286, #469) +# +# The advisory-lock sidecar for a mutable data file. Two claims are load-bearing +# and graded below: WHERE it lives (under the state root, never beside the data +# file, because the ledger is tracked and the engine stages with `git add -A`) +# and WHAT it is keyed on (the resolved path, so every spelling of one file +# contends on one lock). + + +def test_lock_path_for_keys_on_the_resolved_path(tmp_path): + """Two spellings of one ledger get one lock; two ledgers get two. + + A lock keyed on the spelling excludes nobody: the run reaching the ledger by + its `.bmad-loop` relative path and the sweep reaching it through an absolute + or dot-dot spelling would take different sidecars and interleave exactly as + they do today, with the fix installed and inert. + + Also grades the placement, which is the deliberate deviation from #286's own + proposal of a `deferred-work.md.lock` sibling: the ledger is tracked by + design and `verify.commit_story`/`finalize_commit` stage with `git add -A`, + so a sibling would ride into the engine's own commits. + + Ablation: digest `data_path` instead of `data_path.resolve()` and the + dot-dot row fails — one file, two locks.""" + artifacts = tmp_path / "artifacts" + artifacts.mkdir() + ledger = artifacts / "deferred-work.md" + ledger.write_text("# Deferred Work\n", encoding="utf-8") + + direct = runs.lock_path_for(ledger) + # A dot-dot spelling: `str()` keeps it verbatim, so only resolution folds it + dotted = runs.lock_path_for(artifacts / ".." / "artifacts" / "deferred-work.md") + + assert direct == dotted # one file, one lock + assert direct.parent == runs.state_root() / "locks" # never beside the ledger + assert tmp_path not in direct.parents # ...and never inside the project + assert direct.name.endswith("-deferred-work.md.lock") # basename, for humans + + sibling = artifacts / "deferred-work-archive.md" + sibling.write_text("", encoding="utf-8") + assert runs.lock_path_for(sibling) != direct # distinct files, distinct locks + + +def test_lock_path_for_is_pure_and_creates_nothing(tmp_path): + """No mkdir here: `file_lock` mkdirs the lock's parent when it opens it. + + Worth pinning rather than assuming — a helper that provisions the state root + as a side effect of being *asked a question* turns every read-only caller + into a writer, and `lock_path_for` is called from `--json` read models.""" + ledger = tmp_path / "deferred-work.md" + + lock = runs.lock_path_for(ledger) + + assert not lock.exists() + assert not lock.parent.exists() + + +@pytest.mark.skipif(sys.platform == "win32", reason="POSIX symlinks") +def test_lock_path_for_follows_a_symlinked_ledger_to_one_lock(tmp_path): + """The symlink half of the spelling problem — the one no lexical comparison + catches, since the two paths share no component. A project pointed at a + shared external artifact dir through a link contends with the direct + spelling, which is where the real contention is.""" + artifacts = tmp_path / "artifacts" + artifacts.mkdir() + ledger = artifacts / "deferred-work.md" + ledger.write_text("# Deferred Work\n", encoding="utf-8") + link = tmp_path / "link" + link.symlink_to(artifacts, target_is_directory=True) + + assert runs.lock_path_for(link / "deferred-work.md") == runs.lock_path_for(ledger) + + def test_config_digest_is_stamped_under_the_state_root_not_in_the_project(tmp_path): """#498's whole point: the baseline `resume` TRUSTS leaves the tree the driven sessions can write to. From 7981a3e52d775bf2aa959475786ac7e53b65012f Mon Sep 17 00:00:00 2001 From: t Date: Tue, 25 Aug 2026 21:33:28 -0700 Subject: [PATCH 02/12] =?UTF-8?q?feat(deferredwork):=20batched=20locked=20?= =?UTF-8?q?mutation=20primitives=20=E2=80=94=20append=5Fentries,=20mark=5F?= =?UTF-8?q?open=5Fmany,=20record=5Fdecision=20(#286)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- CHANGELOG.md | 8 +- src/bmad_loop/deferredwork.py | 583 +++++++++++++++++++++++----------- tests/test_deferredwork.py | 458 +++++++++++++++++++++++++- 3 files changed, 859 insertions(+), 190 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4e222391..c91c144f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -132,7 +132,13 @@ breaking changes may land in a minor release. is keyed on the resolved path, so every spelling of one file contends on one lock. Nested acquisition raises instead of self-deadlocking, and a lock that cannot be taken fails the write rather than proceeding unlocked. The dev/review session's own ledger writes are - unchanged and still take no lock. + unchanged and still take no lock. Batched primitives collapse the remaining multi-write + sequences into one locked read-modify-write each: `append_entries` files several entries in + one pass (validating every spec before the lock, minting sequential ids, and deduplicating + in-call twins exactly as the loop it replaces did), `mark_open_many` reopens a set of closes, + `record_decision` merges a decision record with its optional closure, and `mark_done_many` + accepts a per-id resolution note. Each is byte-identical to the serial sequence it replaces, + so a caller adopting one changes how many windows it leaves open and nothing else. ### Security diff --git a/src/bmad_loop/deferredwork.py b/src/bmad_loop/deferredwork.py index aa798d72..f2521dba 100644 --- a/src/bmad_loop/deferredwork.py +++ b/src/bmad_loop/deferredwork.py @@ -852,24 +852,34 @@ def _mark_done_many( note: str, *, operation_id: str | None = None, + notes: Sequence[str] | None = None, ) -> list[str]: """Shared atomic implementation for the public close operations. - The whole read->edit->write runs under the cross-process ledger lock - (#286/#469): concurrent mutators — a second run, a sweep, the TUI decision - modal, ``sweep --archive`` — serialize here rather than trading + ONE locked read->edit->write: the whole cycle runs under the cross-process + ledger lock (#286/#469), so concurrent mutators — a second run, a sweep, the + TUI decision modal, ``sweep --archive`` — serialize here rather than trading last-write-wins. Validation stays ABOVE the lock, so a programmer bug reports itself without first waiting on another process. + + ``notes`` supplies a per-id resolution note, positionally paired with + ``dw_ids``; ``note`` is the fallback for every id when it is None. A length + mismatch raises before any I/O rather than closing a prefix under the wrong + evidence — the pairing is positional, so a short list is a caller bug that + would otherwise mis-attribute notes silently. """ _require_iso_date(date) + if notes is not None and len(notes) != len(dw_ids): + raise ValueError(f"notes must be one per dw_id: {len(notes)} for {len(dw_ids)} ids") undo_owner = _operation_digest(operation_id) if operation_id is not None else None with ledger_lock(path): if not path.is_file(): return [] text = path.read_text(encoding="utf-8") marked: list[str] = [] - for dw_id in dw_ids: - updated = _apply_done(text, dw_id, date, note, undo_owner=undo_owner) + for index, dw_id in enumerate(dw_ids): + entry_note = note if notes is None else notes[index] + updated = _apply_done(text, dw_id, date, entry_note, undo_owner=undo_owner) if updated is None: continue text = updated @@ -880,11 +890,23 @@ def _mark_done_many( return marked -def mark_done_many(path: Path, dw_ids: Sequence[str], date: str, note: str) -> list[str]: +def mark_done_many( + path: Path, + dw_ids: Sequence[str], + date: str, + note: str, + *, + notes: Sequence[str] | None = None, +) -> list[str]: """Flip every entry in `dw_ids` to `status: done ` + a resolution note, in ONE read and ONE atomic write. Returns the ids actually flipped (missing and already-done ids are skipped), in the order given. + ``notes[i]`` overrides `note` for ``dw_ids[i]`` — the shape a caller closing + several entries under per-entry evidence needs, which otherwise costs one + read-modify-write cycle per id. A length mismatch raises `ValueError` before + any I/O. + All-or-nothing on purpose. A per-id read-modify-write loop leaves marks on disk when it raises partway through several ids — a half-applied closure the caller never gets to journal, so the ledger claims resolutions the run has no @@ -899,7 +921,7 @@ def mark_done_many(path: Path, dw_ids: Sequence[str], date: str, note: str) -> l ``date`` is validated before the ``is_file`` short-circuit so a programmer bug fails the same way whether or not a ledger happens to exist — a guard that only fires when the file is present is one an absent fixture hides.""" - return _mark_done_many(path, dw_ids, date, note) + return _mark_done_many(path, dw_ids, date, note, notes=notes) def mark_done_many_reopenable( @@ -937,113 +959,209 @@ def mark_done(path: Path, dw_id: str, date: str, note: str) -> bool: ) -def mark_open(path: Path, dw_id: str, note: str, operation_id: str) -> bool: - """Undo one close written by :func:`mark_done_many_reopenable`. +def _apply_open(text: str, dw_id: str, note: str, undo_owner: str) -> str | None: + """Undo one reopenable close *within* `text`. None when the entry is missing, + already open, or does not carry this operation's adjacent resolution and + undo-marker lines. + + Pure by construction — text in, text out, no `Path` and no I/O — which is + what keeps :func:`mark_open_many` able to run it several times inside a + single :func:`ledger_lock` hold. A version of this that touched the file + would have to take the lock itself, and the nested acquisition is exactly the + self-deadlock the guard on `ledger_lock` exists to convert into an error. - The entry must still carry the operation's adjacent resolution and undo-marker - lines. A standard or earlier close has no matching marker and cannot be - reopened merely because it reused the same human-readable note. + A standard or earlier close has no matching marker and cannot be reopened + merely because it reused the same human-readable note. A live ``archived:`` stamp is demoted to :data:`_ARCHIVED_BODY_FIELD` rather than dropped: the reopened entry is no longer archived, but the body its close moved out still is, and that line is the only thing a later triage has - to find it with. - - The whole read->edit->write runs under the cross-process ledger lock - (#286/#469): concurrent mutators — a second run, a sweep, the TUI decision - modal, ``sweep --archive`` — serialize here rather than trading - last-write-wins. - """ + to find it with.""" + entry = _find_entry(text, dw_id) + if entry is None or entry.open: + return None + if entry.status_span is None: + # parse_ledger deliberately tolerates status-less entries. This primitive + # is later called from _defer, where an AttributeError would crash the run + # instead of completing the deferral. + return None + status_line = entry.body[entry.status_span[0] : entry.status_span[1]] + try: + _require_canonical_status(entry.status) + except ValueError: + # Only a canonical status written by mark_done is eligible for undo. + # Preserve malformed or human-authored statuses for validation/reporting. + return None + res_m = _MARK_DONE_TAIL_RE.match(entry.body, entry.status_span[1]) + if res_m is None: + return None + if res_m.group(1).strip() != _one_line(note).strip() or res_m.group(2) != undo_owner: + return None + if status_line != f"status: done {res_m.group(3)}": + return None + try: + previous_status_line = bytes.fromhex(res_m.group(4)).decode("utf-8") + except (UnicodeDecodeError, ValueError): + return None + if LINE_BREAK_RE.search(previous_status_line): + return None + previous_status_m = STATUS_RE.fullmatch(previous_status_line) + previous_status = previous_status_m.group(1).strip() if previous_status_m else "" + if not previous_status or previous_status.split()[0] != "open": + return None + start = entry.span[0] + entry.status_span[0] + end = entry.span[0] + res_m.end() + # Demote the entry's live `archived:` stamps along with the close they + # describe, rather than deleting them. A stub's stamp says "this body lives + # in the archive file"; once the close is undone the body is here and the + # line is a lie, and leaving it standing is not merely untidy — status + + # undo tail + stamp is the exact `_STUB_BODY_RE` shape, so the next + # reopenable close reconstitutes a stub `archive_closed` skips forever, + # stranding the entry outside every future archive (#711). + # + # Cutting the line outright strands the entry a second way: a stub keeps + # neither `location:` nor `reason:` (`_PRESERVED_FIELD_RE`), so the stamp is + # the reopened entry's ONLY route back to the body, and triage arrives with + # a heading and nothing to triage (#711 review). Renaming the field keeps + # both properties — the value still narrows to the archive block, an id + # owning several once a re-closure is archived too, while the renamed line + # matches neither `_ARCHIVED_FIELD_RE` nor `_STUB_BODY_RE`, so the entry + # reads as live and re-archives normally. Rehydrating the body here + # instead was the alternative and is worse: several blocks per id is by + # design, so a rollback's reopen would have to guess which one, and a wrong + # guess overwrites live content with a stale body. + # + # Cuts are disjoint (an `^archived:` line cannot start inside the status + # line or its adjacent tail) and applied back-to-front so earlier offsets + # stay valid. + cuts = [(start, end, previous_status_line)] + for cut_start, cut_end in _archived_line_spans(entry): + # Everything after the field name — value, spacing and the terminating + # newline — carries over verbatim; the span starts at the anchor, so + # the first colon is the field's own. + stamp = entry.body[cut_start:cut_end].split(":", 1)[1] + cuts.append( + ( + entry.span[0] + cut_start, + entry.span[0] + cut_end, + f"{_ARCHIVED_BODY_FIELD}{stamp}", + ) + ) + for cut_start, cut_end, replacement in sorted(cuts, reverse=True): + text = text[:cut_start] + replacement + text[cut_end:] + return text + + +def mark_open_many(path: Path, dw_ids: Sequence[str], note: str, operation_id: str) -> list[str]: + """Undo every close in `dw_ids` written by :func:`mark_done_many_reopenable` + under `operation_id`, in ONE read and ONE atomic write. Returns the ids + actually reopened, in the order given; missing and ineligible ids are + skipped, and an entry whose marker does not match this operation is left + exactly as it was. + + ONE locked read->edit->write: the whole cycle runs under the cross-process + ledger lock (#286/#469), so concurrent mutators — a second run, a sweep, the + TUI decision modal, ``sweep --archive`` — serialize here rather than trading + last-write-wins. A per-id loop over :func:`mark_open` would instead take the + lock once per id, leaving a rival writer a window between every pair of + undos in what a rollback needs to be one step. + + Nothing is written when no id was eligible, so a replayed rollback over + already-reopened entries leaves the file untouched rather than rewriting it + byte-for-byte.""" undo_owner = _operation_digest(operation_id) with ledger_lock(path): if not path.is_file(): - return False + return [] text = path.read_text(encoding="utf-8") - entry = _find_entry(text, dw_id) - if entry is None or entry.open: - return False - if entry.status_span is None: - # parse_ledger deliberately tolerates status-less entries. This primitive - # is later called from _defer, where an AttributeError would crash the run - # instead of completing the deferral. - return False - status_line = entry.body[entry.status_span[0] : entry.status_span[1]] - try: - _require_canonical_status(entry.status) - except ValueError: - # Only a canonical status written by mark_done is eligible for undo. - # Preserve malformed or human-authored statuses for validation/reporting. - return False - res_m = _MARK_DONE_TAIL_RE.match(entry.body, entry.status_span[1]) - if res_m is None: - return False - if res_m.group(1).strip() != _one_line(note).strip() or res_m.group(2) != undo_owner: - return False - if status_line != f"status: done {res_m.group(3)}": - return False - try: - previous_status_line = bytes.fromhex(res_m.group(4)).decode("utf-8") - except (UnicodeDecodeError, ValueError): - return False - if LINE_BREAK_RE.search(previous_status_line): - return False - previous_status_m = STATUS_RE.fullmatch(previous_status_line) - previous_status = previous_status_m.group(1).strip() if previous_status_m else "" - if not previous_status or previous_status.split()[0] != "open": - return False - start = entry.span[0] + entry.status_span[0] - end = entry.span[0] + res_m.end() - # Demote the entry's live `archived:` stamps along with the close they - # describe, rather than deleting them. A stub's stamp says "this body lives - # in the archive file"; once the close is undone the body is here and the - # line is a lie, and leaving it standing is not merely untidy — status + - # undo tail + stamp is the exact `_STUB_BODY_RE` shape, so the next - # reopenable close reconstitutes a stub `archive_closed` skips forever, - # stranding the entry outside every future archive (#711). - # - # Cutting the line outright strands the entry a second way: a stub keeps - # neither `location:` nor `reason:` (`_PRESERVED_FIELD_RE`), so the stamp is - # the reopened entry's ONLY route back to the body, and triage arrives with - # a heading and nothing to triage (#711 review). Renaming the field keeps - # both properties — the value still narrows to the archive block, an id - # owning several once a re-closure is archived too, while the renamed line - # matches neither `_ARCHIVED_FIELD_RE` nor `_STUB_BODY_RE`, so the entry - # reads as live and re-archives normally. Rehydrating the body here - # instead was the alternative and is worse: several blocks per id is by - # design, so a rollback's reopen would have to guess which one, and a wrong - # guess overwrites live content with a stale body. - # - # Cuts are disjoint (an `^archived:` line cannot start inside the status - # line or its adjacent tail) and applied back-to-front so earlier offsets - # stay valid. - cuts = [(start, end, previous_status_line)] - for cut_start, cut_end in _archived_line_spans(entry): - # Everything after the field name — value, spacing and the terminating - # newline — carries over verbatim; the span starts at the anchor, so - # the first colon is the field's own. - stamp = entry.body[cut_start:cut_end].split(":", 1)[1] - cuts.append( - ( - entry.span[0] + cut_start, - entry.span[0] + cut_end, - f"{_ARCHIVED_BODY_FIELD}{stamp}", - ) - ) - for cut_start, cut_end, replacement in sorted(cuts, reverse=True): - text = text[:cut_start] + replacement + text[cut_end:] + reopened: list[str] = [] + for dw_id in dw_ids: + updated = _apply_open(text, dw_id, note, undo_owner) + if updated is None: + continue + text = updated + reopened.append(dw_id) + if not reopened: + return [] atomic_write_text(path, text) - return True + return reopened -def append_decision(path: Path, dw_id: str, date: str, label: str, detail: str) -> bool: - """Record a human decision on an entry without changing its status. +def mark_open(path: Path, dw_id: str, note: str, operation_id: str) -> bool: + """Undo one close written by :func:`mark_done_many_reopenable`. + + A one-id wrapper over :func:`mark_open_many`, which is where the lock is + taken and the contract documented. It delegates rather than duplicating the + read->edit->write so that one public call is exactly one acquisition — a + wrapper that took the lock itself and then called the batch would nest, and + `ledger_lock` raises on that rather than deadlocking.""" + return bool(mark_open_many(path, [dw_id], note, operation_id)) + + +def _apply_decision(text: str, dw_id: str, date: str, label: str, detail: str) -> str | None: + """Insert one `decision: