fix(sweep): diagnose non-completed triage and migration sessions (#752)

CHANGELOG.md

@@ -65,6 +65,11 @@ breaking changes may land in a minor release.
   labelled launch override or policy snapshot, plus its selector; unreadable or
   tampered `sweep.json` reports `unverifiable` (#815).
 
+- Diagnose non-completed sweep triage and migration sessions: journal and escalate
+  whether `result.json` is missing, malformed or valid, and which of the attempt's
+  hook events arrived on either channel (not applicable for a hookless adapter
+  such as opencode-http); routing is unchanged (#752).
+
 - Replace stale installed relay hooks when a project moves between Windows and POSIX.
 
 - Report stale or unverifiable Codex hook trust in `validate` and `probe-adapter`

src/bmad_loop/adapters/generic.py

@@ -29,6 +29,7 @@
 import hashlib
 import json
 import shlex
+import stat
 import time
 from collections.abc import Callable
 from dataclasses import dataclass
@@ -244,6 +245,43 @@ class _IdleTracker:
 )
 
 
+def _result_path(tasks_dir: Path, task_id: str) -> Path:
+    """Where THIS task's result.json lives under an adapter's ``tasks_dir``."""
+    return tasks_dir / task_id / "result.json"
+
+
+def load_result_document(tasks_dir: Path, task_id: str) -> dict | None:
+    """Read one task's skill-written result document exactly as the completion
+    read-back does: ``None`` when no regular file is there; ``OSError``,
+    ``ValueError`` (unparseable JSON, a non-object top level, an unencodable
+    string) or ``RecursionError`` for a present document the read-back refuses.
+    `_ResultFileMixin._read_result` folds every raise into ``None``; the sweep
+    session-failure diagnostic (#752) keeps "missing" and "malformed" apart."""
+    path = _result_path(tasks_dir, task_id)
+    # `stat()` + `S_ISREG`, not `is_file()` (the DW-224 shape): 3.14's `is_file()`
+    # swallows a metadata fault as False, which would report a refused document
+    # as absent; 3.11-3.13 raise. Only absence and a non-directory component are
+    # ``None`` on every runtime — any other fault raises as a present refusal.
+    try:
+        mode = path.stat().st_mode
+    except (FileNotFoundError, NotADirectoryError):
+        return None
+    if not stat.S_ISREG(mode):
+        return None
+    data = json.loads(path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError(f"result.json is not a JSON object: {type(data).__name__}")
+    # 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")
+    return data
+
+
 class _ResultFileMixin:
     """Result-file read-back and verdict finalization: acquire the
     skill-written result dict and fold it into the session's final
@@ -549,7 +587,7 @@ def _final(
         )
 
     def _result_path(self, task_id: str) -> Path:
-        return self.tasks_dir / task_id / "result.json"
+        return _result_path(self.tasks_dir, task_id)
 
     def _append_diag_jsonl(self, task_id: str, filename: str, payload: dict) -> None:
         """Append ``payload`` as one JSON line to ``tasks/<task_id>/<filename>``.
@@ -605,24 +643,10 @@ def _write_heartbeat(self, task_id: str, payload: dict) -> None:
             pass
 
     def _read_result(self, task_id: str) -> dict | None:
-        path = self._result_path(task_id)
         try:
-            if not path.is_file():
-                return None
-            data = json.loads(path.read_text(encoding="utf-8"))
-            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")
+            return load_result_document(self.tasks_dir, task_id)
         except (OSError, ValueError, RecursionError):
             return None
-        return data
 
     def _await_result(self, task_id: str, grace_s: float = RESULT_GRACE_S) -> dict | None:
         deadline = time.monotonic() + grace_s

src/bmad_loop/diagnostics.py

@@ -374,6 +374,13 @@
         # and spec filename. Drop rather than create a second spec correlation;
         # the fallback redacts it only by virtue of its current separators.
         "stashed_to",
+        # A non-completed sweep triage/migration session's post-mortem (#752):
+        # a nested mapping carrying the session task id, validation errors that
+        # quote ledger ids and titles, and exception text naming host paths. The
+        # verdicts are read from the operator's own journal; nothing in it is
+        # needed in a shared dump that `session_status` and `errors` do not
+        # already say, so presence-only rather than a nested routing table.
+        "diagnostic",
         # An overloaded path spelling used for a generated sweep intent and for
         # isolated worktrees. None of those host/customer paths adds useful
         # correlation beyond the record's story key, so every kind gets the same

src/bmad_loop/signals.py

@@ -35,6 +35,69 @@ class HookEvent:
     path: Path
 
 
+def _event_dirs(events_dir: Path, legacy_dir: Path | None) -> list[Path]:
+    """Primary first, then the legacy dir when there is a distinct one. The
+    equality check keeps a caller that passes the same path twice from
+    double-scanning; it cannot produce duplicate events either way (the
+    watcher's ``_consumed`` key would repeat), but the second scan would be
+    pure waste."""
+    if legacy_dir is None or legacy_dir == events_dir:
+        return [events_dir]
+    return [events_dir, legacy_dir]
+
+
+def _parse_event(entry: Path) -> HookEvent | None:
+    """One event file, or ``None`` when it is not a well-formed event."""
+    try:
+        data = json.loads(entry.read_text(encoding="utf-8"))
+    except (json.JSONDecodeError, OSError):
+        return None
+    if not isinstance(data, dict) or "event" not in data or "task_id" not in data:
+        return None
+    return HookEvent(
+        ts=int(data.get("ts", 0)),
+        event=str(data["event"]),
+        task_id=str(data["task_id"]),
+        session_id=data.get("session_id"),
+        transcript_path=data.get("transcript_path"),
+        path=entry,
+    )
+
+
+def is_session_event(event: HookEvent, task_id: str, since_ns: int = 0) -> bool:
+    """The one rule that ties a hook event to a session attempt. ``task_id`` is
+    the session's own id, which folds in the attempt number and generation
+    (``engine._session_task_id``), so another attempt's events never match;
+    ``since_ns`` is that attempt's launch floor, which drops a stale event a
+    resumed run left under the same re-minted id (see ``SignalWatcher.wait_for``)."""
+    return event.task_id == task_id and (not since_ns or event.ts >= since_ns)
+
+
+def session_events(
+    events_dir: Path, legacy_dir: Path | None, task_id: str, since_ns: int = 0
+) -> list[HookEvent]:
+    """Every event on disk for one session attempt, across both channels, oldest
+    first — a read-only snapshot for post-mortem diagnosis (#752), never a
+    completion signal. Unlike ``SignalWatcher.poll`` it creates nothing and
+    consumes nothing, so it sees events the watcher already delivered. A missing
+    directory, primary included, reads as empty (a run whose relay never fired
+    may never have had one made); any other ``OSError`` raises to the caller."""
+    events: list[HookEvent] = []
+    for directory in _event_dirs(events_dir, legacy_dir):
+        try:
+            entries = list(directory.iterdir())
+        except FileNotFoundError:
+            continue
+        for entry in entries:
+            if entry.suffix != ".json":
+                continue
+            event = _parse_event(entry)
+            if event is not None and is_session_event(event, task_id, since_ns):
+                events.append(event)
+    events.sort(key=lambda e: e.ts)
+    return events
+
+
 class SignalWatcher:
     """Poll one or two event directories for a run's hook events.
 
@@ -65,13 +128,7 @@ def __init__(self, events_dir: Path, legacy_dir: Path | None = None):
         events_dir.mkdir(parents=True, exist_ok=True)
 
     def _dirs(self) -> list[Path]:
-        """Primary first, then the legacy dir when there is a distinct one. The
-        equality check keeps a caller that passes the same path twice from
-        double-scanning; it cannot produce duplicate events either way (the
-        ``_consumed`` key would repeat), but the second scan would be pure waste."""
-        if self.legacy_dir is None or self.legacy_dir == self.events_dir:
-            return [self.events_dir]
-        return [self.events_dir, self.legacy_dir]
+        return _event_dirs(self.events_dir, self.legacy_dir)
 
     def poll(self) -> list[HookEvent]:
         """Return new, well-formed events since the last poll, oldest first.
@@ -98,22 +155,9 @@ def poll(self) -> list[HookEvent]:
                 if key in self._consumed or entry.suffix != ".json":
                     continue
                 self._consumed.add(key)
-                try:
-                    data = json.loads(entry.read_text(encoding="utf-8"))
-                except (json.JSONDecodeError, OSError):
-                    continue
-                if not isinstance(data, dict) or "event" not in data or "task_id" not in data:
-                    continue
-                events.append(
-                    HookEvent(
-                        ts=int(data.get("ts", 0)),
-                        event=str(data["event"]),
-                        task_id=str(data["task_id"]),
-                        session_id=data.get("session_id"),
-                        transcript_path=data.get("transcript_path"),
-                        path=entry,
-                    )
-                )
+                event = _parse_event(entry)
+                if event is not None:
+                    events.append(event)
         events.sort(key=lambda e: e.ts)
         return events
 
@@ -153,7 +197,7 @@ def wait_for(
             if since_ns:
                 self._pending = [e for e in self._pending if e.ts >= since_ns]
             for i, event in enumerate(self._pending):
-                if event.task_id == task_id and event.event in kinds:
+                if is_session_event(event, task_id, since_ns) and event.event in kinds:
                     return self._pending.pop(i)
             if clock() >= deadline:
                 return None