fix(verify,diagnostics): dw3-dw5 ledger sweeps (DW-40 … DW-94)

CHANGELOG.md

@@ -85,6 +85,13 @@ breaking changes may land in a minor release.
 
 ### Changed
 
+- **`bmad-loop diagnose --json` reports `schema_version: 4`.** Journal `path` values
+  become `path_present`; stale-restore and merge filename lists become counts.
+
+- **Sanitize remaining diagnostic journal identifiers.** Commit residue and sentinel
+  names are aliased explicitly, excluded and merge filenames are counted, overloaded
+  paths are presence-only, and derived fields win same-named raw-field collisions.
+
 - **Remove the unused whole-artifact-folder exclusion helper** (DW-15). Proof-of-work
   exclusions remain file-granular and rollback protection keeps its workspace-rooted
   path derivation.
@@ -249,6 +256,14 @@ breaking changes may land in a minor release.
 
 ### Fixed
 
+- **The TUI's re-arm declines a contended run instead of waiting for it.** The
+  gesture runs on Textual's message loop, so taking the run's state lock blocking
+  froze the whole dashboard for as long as a rival held it — unbounded on POSIX,
+  where `fcntl.flock` never times out, and a rival `resume` holds it across a git
+  preflight bounded only by `[limits] git_timeout_s`. It now acquires without
+  waiting and toasts the contention; the post-lock liveness re-check refused that
+  re-arm anyway.
+
 - **An orphaned mount's reclaim no longer destroys the orchestrator's own artifacts.** The
   pre-reclaim snapshot drew its untracked candidates from `untracked_files`, which excludes
   ignored paths by contract, and the deferred-work ledger, sprint board and bound spec are
@@ -263,6 +278,45 @@ breaking changes may land in a minor release.
   unrecoverably, on a gitignored ledger. The mark and append legs now re-anchor only when
   the preimage they wrote over is still the bytes the run last claimed; the restore skips
   and journals `ledger-restore-skipped-diverged`.
+- **`bmad-loop clean` no longer aborts the whole sweep when one run's state lock is
+  held.** A busy run used to end the invocation, so later candidates went unprocessed and
+  runs already reclaimed vanished from a report only emitted after the loop. `clean` now
+  takes each run's lock without waiting and records the contended run like the lifecycle
+  races beside it — `trimmed` if anything reached it, else `protected` — then continues.
+
+- **`bmad-loop stop` no longer retries forever on an engine whose identity cannot be
+  read.** The rival-engine compare under the state lock used the local pid after every
+  declining path had cleared it, so an unchanged pid file with `"unknown"` liveness read
+  as a freshly published rival on every attempt. The compare now uses the pid file as
+  recorded; a rival is a changed file, nothing else.
+
+- **The OpenCode adapter validates every task artifact it will write.** It refused a
+  redirected `messages.json` but not a symlinked, hardlinked or special `heartbeat.json`,
+  `resultless-stops.jsonl` or `session-lifecycle.jsonl`, which the inherited result-file
+  mixin writes; the three names now live in one `RESULT_FILE_ARTIFACTS` tuple both adapters
+  validate.
+- Serialize run deletion/archive against resume (DW-94), refusing a newly live
+  engine under the per-run lock and preventing a waiting resume from recreating a
+  run cleanup already removed.
+
+- Serialize every run-state writer and control read-modify-write transaction with one canonical per-run advisory lock (DW-93).
+
+- Let interactive resolve present `paused_reason` when watermark filtering leaves no newer
+  recorded escalation detail, without recovering or inventing an escalation (DW-91).
+
+- Reject malformed session escalation/result artifacts and non-finite resolve JSON
+  (DW-86, DW-89).
+
+- **Confine built-in adapter task directories** (DW-74), refusing unsafe task ids and
+  symlink- or junction-redirected task directories before prompt, artifact, log, or
+  transport side effects.
+
+- **Treat embedded-NUL verify commands and working directories as typed environment
+  faults** (DW-53, DW-54), while documenting that stream retention degrades but
+  journal record writes remain fail-loud.
+
+- Make successful escalation re-arms return authoritative ordered notices and a resume-hold
+  verdict, so a corrupt journal cannot hide a persisted hold from the CLI or TUI gesture.
 - **An accepted spec reached through a link out of the unit worktree no longer counts as
   delivered.** The pre-dispatch check asked only whether a file existed at the mounted path,
   and that probe follows symlinks, so a spec directory the checkout carries as a link

docs/adapter-authoring-guide.md

@@ -581,7 +581,17 @@ Three frozen dataclasses cross the seam:
 
 Required (abstract):
 
-- `start_session(spec) -> SessionHandle` — launch the session.
+- `start_session(spec) -> SessionHandle` — launch the session. An adapter that
+  persists the standard `tasks/<id>/` directory must reset its shared cycle
+  artifacts when an id is reused: after creating the task directory and before
+  launching the session, remove every file named by
+  `journal.TASK_CYCLE_ARTIFACTS` (shared artifacts: [`result.json`,
+  `escalation.json`]).
+  Use missing-safe deletion; a missing artifact is a normal no-op and must not
+  make startup fail. This tuple covers only artifacts shared across adapters and
+  readers. Adapter-private breadcrumbs such as `heartbeat.json`,
+  `resultless-stops.jsonl`, `session-lifecycle.jsonl`, and `messages.json` remain
+  outside the shared cleanup contract and are managed by their owning adapter.
 - `wait_for_completion(handle, spec) -> SessionResult` — block until the session
   ends (or stalls/times out), then report status. Poll
   `runs.read_stop_request_mode(run_dir) == "hard"` on both sides of the loop's

docs/testing.md

@@ -115,6 +115,35 @@ test:
   review sessions do and the orchestrator re-verifies; the bundle twins write none, because
   bundles have no sprint-status entry.)
 
+Divergent-root tests choose a topology for the distinction they need to prove:
+
+- **Default** (`project`) keeps `project == repo_root`. Use it for ordinary sandbox behavior,
+  never for an assertion that claims to distinguish those roots.
+- **Sibling** places `repo_root` beside or otherwise outside the BMAD project. Lower-level seam
+  rows may hand-build this shape (`test_verify.py::_repo_root_override`); caller/config coverage
+  writes it with `write_repo_root_override` and reloads it through `load_paths`. Use it to prove
+  that code commands run in the configured root while artifact reads stay in the project.
+  Artifact-derived excludes correctly collapse to `()` from the disjoint code tree; a non-empty
+  project-relative spelling may still match nothing there, so this shape cannot grade pathspec
+  selection by outcome alone.
+- **Nested monorepo** (`nested_repo_root_paths`) puts the BMAD project at `<repo>/app`, writes
+  and commits `_bmad/bmm/config.yaml`, and returns `load_paths(app)`. Use it when both right-
+  and wrong-root pathspecs must be non-empty and separable: the correct value carries `app/`,
+  while the wrong value can select a plausible outer-tree decoy. The helper accepts alias
+  spellings but returns canonical paths and commits every seed/config file, so fixture residue
+  cannot masquerade as session work.
+- **Isolated worktree** adds a third live root. Use a real relative command against a marker
+  created only in the mounted unit worktree; assert the main checkout lacks it, the command
+  record passed, and classification received the same mounted cwd. Keep the marker under a
+  gitignored generated-state path so it cannot merge back as proof of work.
+
+Every divergent-root row guards its premise before its outcome: compare resolved roots, assert
+the nested parent relation when nesting matters, and for cwd tests plant/probe both directions
+(`plant_root_markers`, `REPO_ROOT_MARKER_CMD`, `PROJECT_MARKER_CMD`). A positive marker identifies
+the intended root; the opposite-root marker rules out the tempting alternative. When a wrong
+root could still name a real path, create that decoy and assert the selected value rather than
+asserting only that some path is absent or a gate refused.
+
 **`MockAdapter` is production code** — `src/bmad_loop/adapters/mock.py`, shipped in the wheel,
 scripted with a list of `SessionResult`s or `callable(spec) -> SessionResult` effects. It is
 not reachable from configuration (no `mock` profile exists; `runsetup.make_adapters` builds

src/bmad_loop/adapters/base.py

@@ -14,12 +14,104 @@
 
 from __future__ import annotations
 
+import stat
 from abc import ABC, abstractmethod
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
 
 from ..model import TokenUsage
+from ..platform_util import is_link_like, safe_segment
+
+
+class AdapterTaskDirectoryError(ValueError):
+    """A built-in adapter refused an unsafe or redirected task directory."""
+
+
+def validated_task_directory(tasks_dir: Path, task_id: str) -> Path:
+    """Return ``tasks/<task_id>`` only when its authored name is confined.
+
+    Validation is deliberately identity-based rather than sanitizing: callers use
+    ``task_id`` for handles, environment, logs, and artifacts, so rewriting it here
+    would split one session across multiple identities.  The link-like check covers
+    both symlinks and Windows directory junctions through ``platform_util``.
+
+    This is a pre-write boundary, not descriptor-anchored I/O; callers must invoke
+    it before any operation derived from the task id.
+    """
+    if safe_segment(task_id) != task_id:
+        raise AdapterTaskDirectoryError(
+            f"unsafe adapter task id {task_id!r}: expected one clean path segment"
+        )
+
+    if is_link_like(tasks_dir):
+        raise AdapterTaskDirectoryError(
+            f"adapter tasks directory is a symlink or junction: {tasks_dir}"
+        )
+
+    task_dir = tasks_dir / task_id
+    if is_link_like(task_dir):
+        raise AdapterTaskDirectoryError(
+            f"adapter task directory is a symlink or junction: {task_dir}"
+        )
+    return task_dir
+
+
+def validate_adapter_artifact_paths(root_dir: Path, paths: tuple[Path, ...]) -> None:
+    """Refuse redirecting or special standing entries before adapter writes.
+
+    A regular file with one link is the only existing leaf an adapter may open in
+    place.  Symlinks, junctions, hardlinks, FIFOs, and devices can redirect or
+    block a later write; callers provide every leaf they will write during the
+    session and invoke this boundary before mutating any task or log artifact.
+    """
+    if is_link_like(root_dir):
+        raise AdapterTaskDirectoryError(
+            f"adapter artifact directory is a symlink or junction: {root_dir}"
+        )
+
+    for path in paths:
+        if is_link_like(path):
+            raise AdapterTaskDirectoryError(f"adapter artifact is a symlink or junction: {path}")
+        try:
+            entry = path.lstat()
+        except FileNotFoundError:
+            continue
+        except OSError as exc:
+            raise AdapterTaskDirectoryError(
+                f"cannot inspect adapter artifact before writing: {path}"
+            ) from exc
+        if not stat.S_ISREG(entry.st_mode) or entry.st_nlink != 1:
+            raise AdapterTaskDirectoryError(
+                f"adapter artifact is special or multiply linked: {path}"
+            )
+
+
+def reset_task_prompt(task_dir: Path, prompt: str) -> None:
+    """Write ``prompt.txt`` without following a redirecting filesystem entry.
+
+    A normal single-link file is truncated in place so its inode and metadata keep
+    the clean-session behavior.  A symlink, hardlink, FIFO, or device is unlinked
+    first so the replacement is an ordinary file and no outside target is touched.
+    """
+    prompt_path = task_dir / "prompt.txt"
+    try:
+        entry = prompt_path.lstat()
+    except FileNotFoundError:
+        pass
+    except OSError as exc:
+        raise AdapterTaskDirectoryError(
+            f"cannot inspect adapter prompt before writing: {prompt_path}"
+        ) from exc
+    else:
+        if not stat.S_ISREG(entry.st_mode) or entry.st_nlink != 1:
+            try:
+                prompt_path.unlink()
+            except OSError as exc:
+                raise AdapterTaskDirectoryError(
+                    f"cannot replace unsafe adapter prompt: {prompt_path}"
+                ) from exc
+    prompt_path.write_text(prompt + "\n", encoding="utf-8")
 
 
 @dataclass(frozen=True)

src/bmad_loop/adapters/generic.py

@@ -21,6 +21,7 @@
 
 from __future__ import annotations
 
+import copy
 import enum
 import hashlib
 import json
@@ -39,7 +40,16 @@
 from ..signals import SignalWatcher
 from ..tokens import read_usage as tally_usage
 from ..verify import read_frontmatter, status_of
-from .base import CodingCLIAdapter, SessionHandle, SessionResult, SessionSpec, SpecSnapshot
+from .base import (
+    CodingCLIAdapter,
+    SessionHandle,
+    SessionResult,
+    SessionSpec,
+    SpecSnapshot,
+    reset_task_prompt,
+    validate_adapter_artifact_paths,
+    validated_task_directory,
+)
 
 # Re-exported for importers that predate the env_fault module split (#194 landed
 # these names on this module); the definitions now live in .env_fault. The
@@ -171,6 +181,21 @@ class _SnapVerdict(enum.Enum):
 )
 
 
+# Every task-directory leaf `_ResultFileMixin` writes during a session, beyond the
+# cycle artifacts in `journal.TASK_CYCLE_ARTIFACTS` and the prompt. Both adapters
+# that inherit the mixin hand this tuple to `validate_adapter_artifact_paths`
+# before their first write: a reused task directory carrying a symlink, hardlink,
+# FIFO or device under one of these names would otherwise have the heartbeat
+# overwrite truncate a linked external file, or a breadcrumb append block on or
+# redirect into it. One tuple, so a fourth mixin write cannot reach one adapter's
+# validation and miss the other's.
+RESULT_FILE_ARTIFACTS: tuple[str, ...] = (
+    "heartbeat.json",
+    "resultless-stops.jsonl",
+    "session-lifecycle.jsonl",
+)
+
+
 class _ResultFileMixin:
     """Result-file read-back and verdict finalization: acquire the
     skill-written result dict and fold it into the session's final
@@ -411,13 +436,23 @@ def _write_heartbeat(self, task_id: str, payload: dict) -> None:
 
     def _read_result(self, task_id: str) -> dict | None:
         path = self._result_path(task_id)
-        if not path.is_file():
-            return None
         try:
+            if not path.is_file():
+                return None
             data = json.loads(path.read_text(encoding="utf-8"))
-        except (json.JSONDecodeError, OSError):
+            if not isinstance(data, dict):
+                return None
+            # Plugin HookContext makes this same defensive copy before exposing
+            # result data, so reject a shape that would recurse there while the
+            # artifact is still inside the shared observation boundary.
+            copy.deepcopy(data)
+            # JSON accepts escaped lone surrogates, but the default ATTENTION
+            # sink writes reasons as UTF-8. Validate every parsed string without
+            # imposing stricter numeric semantics on completed session results.
+            json.dumps(data, ensure_ascii=False).encode("utf-8")
+        except (OSError, ValueError, RecursionError):
             return None
-        return data if isinstance(data, dict) else None
+        return data
 
     def _await_result(self, task_id: str, grace_s: float = RESULT_GRACE_S) -> dict | None:
         deadline = time.monotonic() + grace_s
@@ -538,9 +573,17 @@ def build_command(self, spec: SessionSpec) -> str:
     # --------------------------------------------------------------- adapter
 
     def start_session(self, spec: SessionSpec) -> SessionHandle:
-        task_dir = self.tasks_dir / spec.task_id
+        task_dir = validated_task_directory(self.tasks_dir, spec.task_id)
+        validate_adapter_artifact_paths(
+            task_dir,
+            tuple(task_dir / name for name in RESULT_FILE_ARTIFACTS),
+        )
+        validate_adapter_artifact_paths(
+            self.logs_dir,
+            (self.logs_dir / f"{spec.task_id}.log",),
+        )
         task_dir.mkdir(parents=True, exist_ok=True)
-        (task_dir / "prompt.txt").write_text(spec.prompt + "\n", encoding="utf-8")
+        reset_task_prompt(task_dir, spec.prompt)
         # Task ids are supplied by the caller, so defensively reset cycle-scoped
         # outputs if one is reused. A silent session must not inherit a stale result.
         # The list is `journal.TASK_CYCLE_ARTIFACTS` rather than two literals here:

src/bmad_loop/adapters/opencode_http.py

@@ -150,12 +150,21 @@
 from ..model import TokenUsage
 from ..policy import Policy
 from ..process_host import ProcessHostError, get_process_host
-from .base import CodingCLIAdapter, SessionHandle, SessionResult, SessionSpec
+from .base import (
+    CodingCLIAdapter,
+    SessionHandle,
+    SessionResult,
+    SessionSpec,
+    reset_task_prompt,
+    validate_adapter_artifact_paths,
+    validated_task_directory,
+)
 from .env_fault import EnvFaultMixin
 from .generic import (
     BUDGET_NUDGE_TEXT,
     HEARTBEAT_INTERVAL_S,
     NUDGE_TEXT,
+    RESULT_FILE_ARTIFACTS,
     STALL_NUDGE_TEXT,
     _DevSynthesisMixin,
     _ResultFileMixin,
@@ -619,9 +628,23 @@ def _await_healthy(self, sess: _ServerSession) -> bool:
     # -------------------------------------------------------------- adapter
 
     def start_session(self, spec: SessionSpec) -> SessionHandle:
-        task_dir = self.tasks_dir / spec.task_id
+        task_dir = validated_task_directory(self.tasks_dir, spec.task_id)
+        # `messages.json` is this transport's own; the rest are the inherited
+        # `_ResultFileMixin`'s writes, validated here for the same reason
+        # GenericAdapter validates them — see `RESULT_FILE_ARTIFACTS`.
+        validate_adapter_artifact_paths(
+            task_dir,
+            (task_dir / "messages.json", *(task_dir / name for name in RESULT_FILE_ARTIFACTS)),
+        )
+        log_paths = [
+            self.logs_dir / f"{spec.task_id}.log",
+            self.logs_dir / f"{spec.task_id}.server.out",
+        ]
+        if self.sse_trace:
+            log_paths.append(self.logs_dir / f"{spec.task_id}.sse.jsonl")
+        validate_adapter_artifact_paths(self.logs_dir, tuple(log_paths))
         task_dir.mkdir(parents=True, exist_ok=True)
-        (task_dir / "prompt.txt").write_text(spec.prompt + "\n", encoding="utf-8")
+        reset_task_prompt(task_dir, spec.prompt)
         # Task ids are supplied by the caller, so defensively reset cycle-scoped
         # outputs if one is reused. A silent session must not inherit a stale result.
         # Iterating `journal.TASK_CYCLE_ARTIFACTS` is what makes the parity with