From aac834fa59482807d5e9168f796cb1c74d49b410 Mon Sep 17 00:00:00 2001
From: t <t@t>
Date: Tue, 22 Sep 2026 15:17:06 -0700
Subject: [PATCH 1/3] Diagnose non-completed sweep triage and migration
 sessions (#752)

On a triage or migration session that ends non-completed, journal (and on
escalation, name) whether its result.json is missing, malformed or valid
against the leg's own validator, and which of THIS attempt's hook events
arrived across the primary and legacy channels. Diagnosis only: a valid
artifact is never accepted and routing is unchanged.

Shares the adapter's result.json reader (load_result_document) and the
watcher's event correlation (signals.is_session_event / session_events)
rather than re-spelling either.
---
 CHANGELOG.md                      |   4 +
 src/bmad_loop/adapters/generic.py |  47 ++++--
 src/bmad_loop/diagnostics.py      |   7 +
 src/bmad_loop/signals.py          |  92 ++++++++---
 src/bmad_loop/sweep.py            | 142 +++++++++++++++-
 tests/test_signals.py             |  55 ++++++-
 tests/test_sweep.py               | 259 ++++++++++++++++++++++++++++++
 7 files changed, 562 insertions(+), 44 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index b54603a43..927758722 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -65,6 +65,10 @@ 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; 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`
diff --git a/src/bmad_loop/adapters/generic.py b/src/bmad_loop/adapters/generic.py
index 650cbc77d..56f3fc536 100644
--- a/src/bmad_loop/adapters/generic.py
+++ b/src/bmad_loop/adapters/generic.py
@@ -244,6 +244,35 @@ 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)
+    if not path.is_file():
+        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 +578,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 +634,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
diff --git a/src/bmad_loop/diagnostics.py b/src/bmad_loop/diagnostics.py
index 0809fc1d4..e49c06d7d 100644
--- a/src/bmad_loop/diagnostics.py
+++ b/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
diff --git a/src/bmad_loop/signals.py b/src/bmad_loop/signals.py
index 0c6c33b57..64c482262 100644
--- a/src/bmad_loop/signals.py
+++ b/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
diff --git a/src/bmad_loop/sweep.py b/src/bmad_loop/sweep.py
index 493d76dbd..fc2fb3118 100644
--- a/src/bmad_loop/sweep.py
+++ b/src/bmad_loop/sweep.py
@@ -11,16 +11,19 @@
 
 from __future__ import annotations
 
+import copy
 import json
 import os
 import re
 import stat
+import time
 import unicodedata
 from dataclasses import dataclass, replace
 from pathlib import Path
 from typing import Any, Callable, Iterable, Literal, NoReturn, TypeVar, assert_never
 
 from . import deferredwork, gates, verify
+from .adapters.generic import load_result_document
 from .engine import (
     Engine,
     RunPaused,
@@ -28,6 +31,7 @@
     _ledger_fault_text,
     _LedgerAnchor,
     _publication_refusal,
+    _session_task_id,
 )
 from .escalation import critical_session_reason, env_fault_pause_reason, session_failure_reason
 from .model import PAUSE_STORY_GATE, Phase, StoryTask, result_mapping
@@ -40,7 +44,8 @@
     path_is_confined,
     safe_segment,
 )
-from .runs import StateRootError, _project_of_run_dir
+from .runs import StateRootError, _project_of_run_dir, events_dir_for
+from .signals import session_events
 from .statemachine import advance
 
 
@@ -1316,6 +1321,46 @@ def resolve_sweep_override(override: _T | None, policy_value: _T) -> _T:
     return override if override is not None else policy_value
 
 
+# Bound on any artifact-derived text a session-failure diagnostic carries (#752):
+# the journal records an error summary, never the document itself.
+_DIAGNOSTIC_TEXT_LIMIT = 400
+
+
+def _bounded(text: str, limit: int = _DIAGNOSTIC_TEXT_LIMIT) -> str:
+    return text if len(text) <= limit else text[: limit - 1] + "…"
+
+
+def _triage_verdict(doc: dict[str, Any], open_now: set[str] | None) -> list[str]:
+    """`validate_triage`'s errors for ``doc`` as the completed path would grade it —
+    bundle names normalized first — on a private copy, so a diagnostic read never
+    repairs the artifact it inspects."""
+    graded = copy.deepcopy(doc)
+    _normalize_bundle_names(graded)
+    return validate_triage(graded, open_now)[1]
+
+
+def _hook_verdict(kinds: set[str]) -> str:
+    if not kinds:
+        return "none"
+    if "Stop" in kinds:
+        return "stop"
+    if "SessionStart" in kinds:
+        return "session-start-without-stop"
+    return "no-session-start-or-stop"
+
+
+def _diagnostic_suffix(diagnostic: dict[str, Any] | None) -> str:
+    """The escalation-text tail for a non-completed session's diagnostic (#752):
+    what was on disk and which hook events arrived, so an operator whose event
+    channel is miswired debugs the channel, not the agent's output."""
+    if diagnostic is None:
+        return ""
+    artifact = str(diagnostic["artifact"])
+    if diagnostic.get("artifact_error"):
+        artifact += f" ({diagnostic['artifact_error']})"
+    return f" [result.json: {artifact}; hook events: {diagnostic['hook_events']}]"
+
+
 class SweepEngine(Engine):
     """Engine variant whose loop processes the deferred-work ledger instead
     of sprint-status. Bundles reuse the inherited story pipeline through the
@@ -3672,6 +3717,7 @@ def _ensure_migration(self, text: str) -> None:
             task.attempt += 1
             advance(task, Phase.TRIAGE_RUNNING)
             self._save()
+            launch_floor_ns = time.time_ns()
             result = self._run_session(
                 task,
                 role="triage",
@@ -3700,8 +3746,14 @@ def _ensure_migration(self, text: str) -> None:
             # anchor's "the session deleted it", distinct from an empty write.
             rewrite = deferredwork.read_for_write(ledger)
             new_text = rewrite if rewrite is not None else ""
+            diagnostic: dict[str, Any] | None = None
             if result.status != "completed":
                 errors = [session_failure_reason("migration", result)]
+                diagnostic = self._session_failure_diagnostic(
+                    task,
+                    launch_floor_ns,
+                    lambda doc: validate_migration(doc, manifest, pre_canonical, new_text),
+                )
             else:
                 errors = validate_migration(result.result_json, manifest, pre_canonical, new_text)
             self.journal.append(
@@ -3711,6 +3763,7 @@ def _ensure_migration(self, text: str) -> None:
                 ok=not errors,
                 errors=errors,
                 env_fault=result.env_fault,
+                diagnostic=diagnostic,
             )
             if result.status != "completed" and result.env_fault:
                 # The migration session's CLI lost its API connection (#194): it did
@@ -3838,7 +3891,10 @@ def _ensure_migration(self, text: str) -> None:
                 )
             if task.attempt >= self.policy.sweep.max_migration_attempts:
                 self._escalate(
-                    task, "migration failed deterministic validation: " + "; ".join(errors)
+                    task,
+                    "migration failed deterministic validation: "
+                    + "; ".join(errors)
+                    + _diagnostic_suffix(diagnostic),
                 )
             feedback = self._write_feedback(
                 task,
@@ -3846,6 +3902,75 @@ def _ensure_migration(self, text: str) -> None:
                 + "\n- ".join(errors),
             )
 
+    def _session_failure_diagnostic(
+        self,
+        task: StoryTask,
+        launch_floor_ns: int,
+        validate: Callable[[dict[str, Any]], list[str]],
+    ) -> dict[str, Any]:
+        """Observe what a NON-completed triage/migration session left behind (#752):
+        its ``result.json`` and the hook events of THIS attempt. Diagnosis only —
+        a valid artifact here is never a completion (sessions complete on Stop or
+        window death alone), and nothing in it changes the attempt's routing.
+
+        ``artifact``: ``missing`` | ``malformed`` (``artifact_error`` names the parse
+        or validation failure, bounded) | ``valid`` | ``unreadable: <reason>``.
+        ``validate`` is the leg's own completed-path validator, handed the parsed
+        document read-only.
+
+        ``hook_events``: ``none`` | ``session-start-without-stop`` | ``stop`` |
+        ``no-session-start-or-stop`` | ``unreadable: <reason>``, counted over both
+        the out-of-tree channel and the legacy in-tree ``<run_dir>/events`` through
+        ``signals.is_session_event`` — this attempt's session id and launch floor,
+        so an earlier healthy attempt's events cannot mask this failure.
+
+        Observation degrades, never raises: this runs on a failure path that must
+        still reach its retry/escalation decision."""
+        # The id `_run_session` just minted for this attempt: the sweep dispatches
+        # both legs as role "triage" with no label.
+        task_id = _session_task_id(task.story_key, "triage", task.attempt, task.generation)
+        diagnostic: dict[str, Any] = {"task_id": task_id}
+        # The adapter's own read-back location, which it clears at launch, so a
+        # document there was written during this attempt.
+        try:
+            doc = load_result_document(self.run_dir / "tasks", task_id)
+        except OSError as exc:
+            diagnostic["artifact"] = f"unreadable: {_bounded(str(exc))}"
+        except (ValueError, RecursionError) as exc:
+            diagnostic["artifact"] = "malformed"
+            diagnostic["artifact_error"] = _bounded(f"{type(exc).__name__}: {exc}")
+        else:
+            if doc is None:
+                diagnostic["artifact"] = "missing"
+            else:
+                try:
+                    errors = validate(doc)
+                except Exception as exc:  # observation degrades; see docstring
+                    diagnostic["artifact"] = (
+                        f"unreadable: validator raised {_bounded(f'{type(exc).__name__}: {exc}')}"
+                    )
+                else:
+                    if errors:
+                        diagnostic["artifact"] = "malformed"
+                        diagnostic["artifact_error"] = _bounded("; ".join(errors))
+                    else:
+                        diagnostic["artifact"] = "valid"
+        try:
+            events = session_events(
+                events_dir_for(self.paths.project, self.run_dir.name),
+                self.run_dir / "events",
+                task_id,
+                launch_floor_ns,
+            )
+        except Exception as exc:  # observation degrades; see docstring
+            diagnostic["hook_events"] = f"unreadable: {_bounded(f'{type(exc).__name__}: {exc}')}"
+        else:
+            kinds = {event.event for event in events}
+            diagnostic["hook_events"] = _hook_verdict(kinds)
+            diagnostic["hook_event_kinds"] = sorted(kinds)
+            diagnostic["hook_event_count"] = len(events)
+        return diagnostic
+
     def _migrate_prompt(self, manifest: Path, feedback: Path | None) -> str:
         prompt = f"/bmad-loop-sweep --migrate {manifest}"
         if feedback is not None:
@@ -3942,6 +4067,7 @@ def _ensure_triage(self, open_now: set[str], cycle: int = 1) -> TriagePlan:
             task.attempt += 1
             advance(task, Phase.TRIAGE_RUNNING)
             self._save()
+            launch_floor_ns = time.time_ns()
             result = self._run_session(
                 task,
                 role="triage",
@@ -3953,8 +4079,12 @@ def _ensure_triage(self, open_now: set[str], cycle: int = 1) -> TriagePlan:
             critical_reason = critical_session_reason("triage", result.result_json)
             if critical_reason is not None:
                 self._escalate(task, critical_reason)
+            diagnostic: dict[str, Any] | None = None
             if result.status != "completed":
                 plan, errors = None, [session_failure_reason("triage", result)]
+                diagnostic = self._session_failure_diagnostic(
+                    task, launch_floor_ns, lambda doc: _triage_verdict(doc, open_now)
+                )
             else:
                 repairs = _normalize_bundle_names(result.result_json)
                 plan, errors = validate_triage(result.result_json, open_now)
@@ -3972,6 +4102,7 @@ def _ensure_triage(self, open_now: set[str], cycle: int = 1) -> TriagePlan:
                 ok=plan is not None,
                 errors=errors,
                 env_fault=result.env_fault,
+                diagnostic=diagnostic,
             )
             if result.status != "completed" and result.env_fault:
                 # transport/API failure (#194): pause rather than charge a triage
@@ -4032,7 +4163,12 @@ def _ensure_triage(self, open_now: set[str], cycle: int = 1) -> TriagePlan:
                 self._emit("post_triage", task)
                 return plan
             if task.attempt >= self.policy.sweep.max_triage_attempts:
-                self._escalate(task, "triage output failed validation: " + "; ".join(errors))
+                self._escalate(
+                    task,
+                    "triage output failed validation: "
+                    + "; ".join(errors)
+                    + _diagnostic_suffix(diagnostic),
+                )
             feedback = self._write_feedback(
                 task,
                 "The triage result.json failed deterministic validation:\n- " + "\n- ".join(errors),
diff --git a/tests/test_signals.py b/tests/test_signals.py
index ce6a76200..173984328 100644
--- a/tests/test_signals.py
+++ b/tests/test_signals.py
@@ -2,7 +2,7 @@
 
 import pytest
 
-from bmad_loop.signals import SignalWatcher
+from bmad_loop.signals import SignalWatcher, is_session_event, session_events
 
 
 def write_event(events_dir, ts, task_id, event, **extra):
@@ -179,3 +179,56 @@ def test_poll_still_raises_when_the_primary_dir_is_gone(tmp_path):
 
     with pytest.raises(OSError):
         watcher.poll()
+
+
+def test_session_events_keeps_only_this_attempt_across_both_channels(tmp_path):
+    """The #752 diagnostic's read: one attempt's events from the primary AND legacy
+    channels, oldest first, excluding another attempt's id and anything stamped
+    before this attempt's launch floor (a resumed run's same-id leftovers).
+
+    Ablation guard: drop the task-id or the floor term from `is_session_event`, or
+    the legacy dir from the scan, and this fails."""
+    primary, legacy = tmp_path / "state" / "events", tmp_path / "run" / "events"
+    write_event(primary, 50, "t-2", "SessionStart")
+    write_event(legacy, 60, "t-2", "Stop")
+    write_event(primary, 55, "t-1", "Stop")  # another attempt, inside the window
+    write_event(legacy, 5, "t-2", "Stop")  # this id, but before the launch floor
+
+    events = session_events(primary, legacy, "t-2", since_ns=10)
+
+    assert [(e.ts, e.event) for e in events] == [(50, "SessionStart"), (60, "Stop")]
+
+
+def test_session_events_is_read_only_and_tolerates_missing_dirs(tmp_path):
+    """A post-mortem read neither creates a channel nor consumes from one: a relay
+    that never fired may have left no directory at all, and a live watcher must
+    still see every event the diagnostic looked at."""
+    primary, legacy = tmp_path / "state" / "events", tmp_path / "run" / "events"
+    assert session_events(primary, legacy, "t1") == []
+    assert not primary.exists() and not legacy.exists()
+
+    watcher = SignalWatcher(primary, legacy)
+    write_event(primary, 1, "t1", "Stop")
+    assert len(session_events(primary, legacy, "t1")) == 1
+    assert watcher.wait_for("t1", {"Stop"}, timeout_s=1) is not None
+
+
+def test_session_events_raises_on_an_unreadable_channel(tmp_path):
+    """Only absence reads as empty; any other fault surfaces for the caller to
+    report as unreadable rather than pass as "no events"."""
+    primary = tmp_path / "events"
+    primary.write_text("not a directory")
+
+    with pytest.raises(OSError):
+        session_events(primary, None, "t1")
+
+
+def test_is_session_event_is_the_rule_wait_for_matches_on(tmp_path):
+    """One correlation rule, shared by the completion wait and the diagnostic."""
+    write_event(tmp_path, 7, "t1", "Stop")
+    (event,) = SignalWatcher(tmp_path).poll()
+
+    assert is_session_event(event, "t1")
+    assert is_session_event(event, "t1", since_ns=7)
+    assert not is_session_event(event, "t1", since_ns=8)
+    assert not is_session_event(event, "t2")
diff --git a/tests/test_sweep.py b/tests/test_sweep.py
index 18e8532b8..27045ce24 100644
--- a/tests/test_sweep.py
+++ b/tests/test_sweep.py
@@ -6,6 +6,7 @@
 import re
 import shutil
 import sys
+import time
 from dataclasses import replace
 from pathlib import Path
 from typing import Literal
@@ -5246,6 +5247,264 @@ def test_migration_lost_session_names_the_mux_in_its_errors(project):
     assert any("multiplexer no longer reports the session" in e for e in dec["errors"])
 
 
+# --- #752: a non-completed triage/migration session is diagnosed, never accepted ---
+
+
+def failed_session_effect(
+    status="timeout",
+    artifact=None,
+    primary_events=(),
+    legacy_events=(),
+    ledger=None,
+    result=None,
+):
+    """A scripted session that ends NON-completed after leaving, where a real one
+    would, the evidence the #752 diagnostic reads: `artifact` (a dict, or raw text
+    for a malformed document) at `<run_dir>/tasks/<task_id>/result.json`, and hook
+    events on the out-of-tree channel (`BMAD_LOOP_EVENTS_DIR`) and/or the legacy
+    in-tree `<run_dir>/events`, correlated exactly as the relay stamps them.
+    `ledger` rewrites the deferred-work ledger first (a migration's work)."""
+
+    def effect(spec):
+        run_dir = Path(spec.env["BMAD_LOOP_RUN_DIR"])
+        task_id = spec.env["BMAD_LOOP_TASK_ID"]
+        if ledger is not None:
+            ledger[0].write_text(ledger[1], encoding="utf-8")
+        if artifact is not None:
+            task_dir = run_dir / "tasks" / task_id
+            task_dir.mkdir(parents=True, exist_ok=True)
+            text = artifact if isinstance(artifact, str) else json.dumps(artifact)
+            (task_dir / "result.json").write_text(text, encoding="utf-8")
+        channels = (
+            (Path(spec.env["BMAD_LOOP_EVENTS_DIR"]), primary_events),
+            (run_dir / "events", legacy_events),
+        )
+        for directory, kinds in channels:
+            for kind in kinds:
+                directory.mkdir(parents=True, exist_ok=True)
+                ts = time.time_ns()
+                payload = {"ts": ts, "event": kind, "task_id": task_id, "session_id": "s"}
+                (directory / f"{ts}-{task_id}-{kind}.json").write_text(json.dumps(payload))
+        return result if result is not None else SessionResult(status=status)
+
+    return effect
+
+
+def _decisions(engine, kind):
+    return [e for e in engine.journal.entries() if e["kind"] == kind]
+
+
+VALID_TRIAGE = triage_result(["DW-1"], skip=[{"id": "DW-1", "reason": "moot"}])
+
+
+@pytest.mark.parametrize(
+    ("artifact", "verdict", "error_fragment"),
+    [
+        (None, "missing", None),
+        ("{not json", "malformed", "JSONDecodeError"),
+        ({"workflow": "something-else"}, "malformed", "workflow"),
+        (VALID_TRIAGE, "valid", None),
+    ],
+)
+def test_non_completed_triage_diagnoses_its_artifact_without_changing_the_outcome(
+    project, artifact, verdict, error_fragment
+):
+    """#752: a timed-out triage whose result.json is missing, malformed, or even fully
+    valid is diagnosed in `triage-decision` and in the escalation text — and routed
+    exactly as today: both attempts spent, then ESCALATED, never accepted.
+
+    Ablation guard (valid row): route a `valid` diagnostic to completion and this
+    fails — the run finishes instead of escalating."""
+    write_ledger(project, {"DW-1": "open"})
+    engine, adapter = make_sweep(
+        project,
+        [
+            failed_session_effect(artifact=artifact, primary_events=["SessionStart"]),
+            failed_session_effect(artifact=artifact, primary_events=["SessionStart"]),
+        ],
+    )
+    summary = engine.run()
+
+    assert summary.paused
+    task = engine.state.tasks["sweep-triage"]
+    assert task.phase == Phase.ESCALATED
+    assert task.attempt == 2
+    assert len(adapter.sessions) == 2
+    assert not (engine.run_dir / "triage.json").exists()  # no plan was cached
+    decisions = _decisions(engine, "triage-decision")
+    assert [d["ok"] for d in decisions] == [False, False]
+    assert all(d["errors"] == ["triage session timeout"] for d in decisions)
+    diag = decisions[-1]["diagnostic"]
+    assert diag["task_id"] == "sweep-triage-triage-2"
+    assert diag["artifact"] == verdict
+    if error_fragment is None:
+        assert "artifact_error" not in diag
+    else:
+        assert error_fragment in diag["artifact_error"]
+    assert f"[result.json: {verdict}" in engine.state.paused_reason
+    assert "hook events: session-start-without-stop]" in engine.state.paused_reason
+
+
+@pytest.mark.parametrize(
+    ("artifact", "verdict", "error_fragment"),
+    [
+        (None, "missing", None),
+        ("[1, 2", "malformed", "JSONDecodeError"),
+        ({"workflow": "deferred-sweep-migrate", "mapping": []}, "malformed", "not mapped"),
+        ("valid", "valid", None),
+    ],
+)
+def test_non_completed_migration_diagnoses_its_artifact_without_changing_the_outcome(
+    project, artifact, verdict, error_fragment
+):
+    """The migration twin: a timed-out session that rewrote the ledger correctly and
+    wrote a validating result.json is still a failed attempt — the rewrite is
+    reset, both attempts spent, ESCALATED — with the diagnostic saying the artifact
+    validated against the very ledger text this attempt left.
+
+    Ablation guard (valid row): route a `valid` diagnostic to completion and this
+    fails."""
+    write_legacy_ledger(project, LEGACY_LEDGER)
+    manifest = legacy_manifest()
+    mapping = [
+        {"key": manifest[0]["key"], "dw_id": "DW-1"},
+        {"key": manifest[1]["key"], "dw_id": "DW-2"},
+    ]
+    doc = migrate_result(mapping) if artifact == "valid" else artifact
+    rewrite = (project.deferred_work, migrated_ledger())
+    engine, adapter = make_sweep(
+        project,
+        [
+            failed_session_effect(artifact=doc, ledger=rewrite),
+            failed_session_effect(artifact=doc, ledger=rewrite),
+        ],
+    )
+    summary = engine.run()
+
+    assert summary.paused
+    task = engine.state.tasks["sweep-migrate"]
+    assert task.phase == Phase.ESCALATED
+    assert task.attempt == 2
+    assert len(adapter.sessions) == 2
+    assert project.deferred_work.read_text(encoding="utf-8") == LEGACY_LEDGER  # reset
+    decisions = _decisions(engine, "migrate-decision")
+    assert [d["ok"] for d in decisions] == [False, False]
+    assert all(d["errors"] == ["migration session timeout"] for d in decisions)
+    diag = decisions[-1]["diagnostic"]
+    assert diag["task_id"] == "sweep-migrate-triage-2"
+    assert diag["artifact"] == verdict
+    if error_fragment is None:
+        assert "artifact_error" not in diag
+    else:
+        assert error_fragment in diag["artifact_error"]
+    assert diag["hook_events"] == "none"
+    assert f"[result.json: {verdict}" in engine.state.paused_reason
+
+
+@pytest.mark.parametrize(
+    ("primary", "verdict", "kinds"),
+    [
+        ((), "none", []),
+        (("SessionStart",), "session-start-without-stop", ["SessionStart"]),
+        (("SessionStart", "Stop"), "stop", ["SessionStart", "Stop"]),
+        (("SessionEnd",), "no-session-start-or-stop", ["SessionEnd"]),
+    ],
+)
+def test_non_completed_triage_reports_hook_evidence_distinctly(project, primary, verdict, kinds):
+    """No events at all (a relay that never fired — #752's incident) and a
+    SessionStart that never reached Stop (a session that wedged) are different
+    faults and are journaled as different verdicts."""
+    write_ledger(project, {"DW-1": "open"})
+    engine, _ = make_sweep(
+        project,
+        [failed_session_effect(primary_events=primary)] * 2,
+    )
+    engine.run()
+
+    diag = _decisions(engine, "triage-decision")[-1]["diagnostic"]
+    assert diag["hook_events"] == verdict
+    assert diag["hook_event_kinds"] == kinds
+    assert diag["hook_event_count"] == len(primary)
+
+
+def test_failed_triage_after_a_healthy_attempt_reports_only_its_own_evidence(project):
+    """Attempt 1 ran cleanly — SessionStart, Stop, a result.json on disk — but its
+    plan failed validation; attempt 2 then timed out having left nothing. The
+    diagnostic must describe attempt 2 alone: no events, no artifact. Attempt 1's
+    evidence sits in the same channels and must not mask the failure.
+
+    Ablation guard: make `signals.session_events` keep every event regardless of
+    session id and launch floor and this fails on `hook_events`."""
+    write_ledger(project, {"DW-1": "open"})
+    bad_plan = {"workflow": "deferred-sweep-triage"}
+    healthy = failed_session_effect(
+        artifact=VALID_TRIAGE,
+        primary_events=["SessionStart", "Stop"],
+        result=SessionResult(status="completed", result_json=bad_plan),
+    )
+    engine, _ = make_sweep(project, [healthy, failed_session_effect()])
+    engine.run()
+
+    first, second = _decisions(engine, "triage-decision")
+    assert first["diagnostic"] is None  # a completed session is graded, not diagnosed
+    diag = second["diagnostic"]
+    assert diag["task_id"] == "sweep-triage-triage-2"
+    assert diag["hook_events"] == "none"
+    assert diag["hook_event_count"] == 0
+    assert diag["artifact"] == "missing"
+
+
+def test_non_completed_triage_counts_legacy_channel_events_for_the_attempt(project):
+    """A project on an older installed relay writes to the legacy in-tree
+    `<run_dir>/events` (see `signals`). Its events belong to the attempt as much as
+    the primary channel's do: SessionStart on the primary and Stop on the legacy
+    channel is one attempt that DID end its turn.
+
+    Ablation guard: drop the legacy dir from the diagnostic scan and this fails."""
+    write_ledger(project, {"DW-1": "open"})
+    effect = failed_session_effect(primary_events=["SessionStart"], legacy_events=["Stop"])
+    engine, _ = make_sweep(project, [effect, effect])
+    engine.run()
+
+    diag = _decisions(engine, "triage-decision")[-1]["diagnostic"]
+    assert diag["hook_events"] == "stop"
+    assert diag["hook_event_count"] == 2
+
+
+def test_session_failure_diagnostic_degrades_to_unreadable_and_keeps_routing(project, monkeypatch):
+    """Observation may degrade, never raise: an artifact read and an event scan that
+    both fail still land a journaled decision and the ordinary escalation."""
+
+    def refuse(*_args, **_kwargs):
+        raise PermissionError("denied")
+
+    monkeypatch.setattr(sweep_mod, "load_result_document", refuse)
+    monkeypatch.setattr(sweep_mod, "session_events", refuse)
+    write_ledger(project, {"DW-1": "open"})
+    engine, adapter = make_sweep(project, [SessionResult(status="timeout")] * 2)
+    engine.run()
+
+    assert engine.state.tasks["sweep-triage"].phase == Phase.ESCALATED
+    assert len(adapter.sessions) == 2
+    diag = _decisions(engine, "triage-decision")[-1]["diagnostic"]
+    assert diag["artifact"] == "unreadable: denied"
+    assert diag["hook_events"].startswith("unreadable: PermissionError")
+
+
+def test_session_failure_diagnostic_bounds_the_artifact_error(project):
+    """The journal carries an error summary, never the document: a malformed
+    artifact whose validation errors run long is cut to the diagnostic bound."""
+    write_ledger(project, {"DW-1": "open"})
+    huge = triage_result(["DW-1"], skip=[{"id": f"DW-{n}", "reason": "x"} for n in range(2, 400)])
+    engine, _ = make_sweep(project, [failed_session_effect(artifact=huge)] * 2)
+    engine.run()
+
+    diag = _decisions(engine, "triage-decision")[-1]["diagnostic"]
+    assert diag["artifact"] == "malformed"
+    assert len(diag["artifact_error"]) <= sweep_mod._DIAGNOSTIC_TEXT_LIMIT
+    assert diag["artifact_error"].endswith("…")
+
+
 def test_migration_session_env_fault_escalates_without_consuming_attempts(project):
     """A migration session whose CLI lost its API connection (#194) escalates on the
     first attempt instead of charging a migration retry; migrate-decision carries

From 676f0238da08fc85fd9333f503c1ece7c6b6eea3 Mon Sep 17 00:00:00 2001
From: t <t@t>
Date: Tue, 22 Sep 2026 21:25:49 -0700
Subject: [PATCH 2/3] Keep a refused result.json probe and env-fault
 escalations diagnosed (#752)

load_result_document probes with stat() + S_ISREG instead of is_file(), so a
metadata refusal raises (diagnosed "unreadable") on every runtime rather than
reading as "missing" on 3.14. The env-fault escalation branches for triage and
migration now append the session-failure diagnostic, as the attempt-cap
escalation already did.
---
 src/bmad_loop/adapters/generic.py | 11 ++++++-
 src/bmad_loop/sweep.py            |  4 +--
 tests/test_generic_tmux.py        | 49 ++++++++++++++++++++++++++-----
 tests/test_sweep.py               |  4 +++
 4 files changed, 57 insertions(+), 11 deletions(-)

diff --git a/src/bmad_loop/adapters/generic.py b/src/bmad_loop/adapters/generic.py
index 56f3fc536..b6735b11e 100644
--- a/src/bmad_loop/adapters/generic.py
+++ b/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
@@ -257,7 +258,15 @@ def load_result_document(tasks_dir: Path, task_id: str) -> dict | None:
     `_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)
-    if not path.is_file():
+    # `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):
diff --git a/src/bmad_loop/sweep.py b/src/bmad_loop/sweep.py
index fc2fb3118..f6970c9cb 100644
--- a/src/bmad_loop/sweep.py
+++ b/src/bmad_loop/sweep.py
@@ -3773,7 +3773,7 @@ def _ensure_migration(self, text: str) -> None:
                 # resume also restores the ledger if the worktree is dirty.
                 self._escalate(
                     task,
-                    env_fault_pause_reason("migration", result),
+                    env_fault_pause_reason("migration", result) + _diagnostic_suffix(diagnostic),
                 )
             if not errors:
                 # This record is the durable proof that validation accepted the
@@ -4110,7 +4110,7 @@ def _ensure_triage(self, open_now: set[str], cycle: int = 1) -> TriagePlan:
                 # ESCALATED-resume above resets task.attempt to 0 (fresh budget).
                 self._escalate(
                     task,
-                    env_fault_pause_reason("triage", result),
+                    env_fault_pause_reason("triage", result) + _diagnostic_suffix(diagnostic),
                 )
             if plan is not None:
                 advance(task, Phase.DONE)
diff --git a/tests/test_generic_tmux.py b/tests/test_generic_tmux.py
index 9ead259c7..cbddc53b9 100644
--- a/tests/test_generic_tmux.py
+++ b/tests/test_generic_tmux.py
@@ -259,21 +259,54 @@ def test_read_result_degrades_a_decoder_recursion_error(tmp_path):
     assert adapter._read_result("t1") is None
 
 
-def test_read_result_degrades_an_unreadable_existence_probe(tmp_path, monkeypatch):
-    adapter = make_adapter(tmp_path)
-    path = adapter._result_path("t1")
-    real_is_file = Path.is_file
+def _refuse_result_metadata(monkeypatch, path):
+    """Refuse `stat()` on ``path`` and pin `Path.is_file` False for it — 3.14's
+    answer to that refusal — so a reader that went back to `is_file()` would see
+    "absent" on every runtime, not only on 3.14."""
+    real_stat, real_is_file = Path.stat, Path.is_file
 
-    def is_file_with_permission_error(candidate):
+    def refused_stat(candidate, *args, **kwargs):
         if candidate == path:
-            raise PermissionError("task directory is not searchable")
-        return real_is_file(candidate)
+            raise PermissionError(13, "task directory is not searchable", str(candidate))
+        return real_stat(candidate, *args, **kwargs)
+
+    def swallowed_is_file(candidate, *args, **kwargs):
+        return False if candidate == path else real_is_file(candidate, *args, **kwargs)
+
+    monkeypatch.setattr(Path, "stat", refused_stat)
+    monkeypatch.setattr(Path, "is_file", swallowed_is_file)
 
-    monkeypatch.setattr(Path, "is_file", is_file_with_permission_error)
+
+def test_read_result_degrades_an_unreadable_existence_probe(tmp_path, monkeypatch):
+    adapter = make_adapter(tmp_path)
+    _refuse_result_metadata(monkeypatch, adapter._result_path("t1"))
 
     assert adapter._read_result("t1") is None
 
 
+def test_load_result_document_raises_a_refused_probe_instead_of_reporting_absence(
+    tmp_path, monkeypatch
+):
+    """`None` is absence alone (#752): the sweep diagnostic reports it as `missing`,
+    so a metadata refusal must raise and read as `unreadable` on every runtime.
+
+    Ablation: restore `if not path.is_file(): return None` and this fails."""
+    adapter = make_adapter(tmp_path)
+    task_dir = adapter.tasks_dir / "t1"
+    task_dir.mkdir(parents=True)
+    assert generic.load_result_document(adapter.tasks_dir, "t1") is None  # absent
+    (task_dir / "result.json").mkdir()
+    assert generic.load_result_document(adapter.tasks_dir, "t1") is None  # not a regular file
+    (task_dir / "result.json").rmdir()
+    (tmp_path / "plain").write_text("x")
+    assert generic.load_result_document(tmp_path / "plain", "t1") is None  # non-dir component
+
+    (task_dir / "result.json").write_text('{"clean": true}')
+    _refuse_result_metadata(monkeypatch, adapter._result_path("t1"))
+    with pytest.raises(PermissionError):
+        generic.load_result_document(adapter.tasks_dir, "t1")
+
+
 def test_read_result_rejects_data_that_plugin_deepcopy_cannot_handle(tmp_path):
     adapter = make_adapter(tmp_path)
     task_dir = adapter.tasks_dir / "t1"
diff --git a/tests/test_sweep.py b/tests/test_sweep.py
index 27045ce24..881ca2536 100644
--- a/tests/test_sweep.py
+++ b/tests/test_sweep.py
@@ -5105,6 +5105,9 @@ def test_triage_session_env_fault_escalates_then_resume_restores_budget(project)
     assert task.attempt == 1  # only the one session — no retry budget spent
     assert "environment fault: triage session timeout" in engine.state.paused_reason
     assert evidence in engine.state.paused_reason
+    # #752: the env-fault escalation carries the attempt's diagnostic too
+    # (ablation: drop `_diagnostic_suffix` from the env-fault `_escalate`).
+    assert "[result.json: missing; hook events: none]" in engine.state.paused_reason
     assert len(adapter.sessions) == 1  # no feedback-retry session
     dec = [e for e in engine.journal.entries() if e["kind"] == "triage-decision"][-1]
     assert dec["env_fault"] is True
@@ -5523,6 +5526,7 @@ def test_migration_session_env_fault_escalates_without_consuming_attempts(projec
     assert task.attempt == 1  # only the one session — no migration retry spent
     assert "environment fault: migration session timeout" in engine.state.paused_reason
     assert evidence in engine.state.paused_reason
+    assert "[result.json: missing; hook events: none]" in engine.state.paused_reason  # #752
     assert len(adapter.sessions) == 1
     dec = [e for e in engine.journal.entries() if e["kind"] == "migrate-decision"][-1]
     assert dec["env_fault"] is True

From 15d8f03717192c972238fea5549def75978b042a Mon Sep 17 00:00:00 2001
From: t <t@t>
Date: Tue, 22 Sep 2026 21:39:16 -0700
Subject: [PATCH 3/3] Report hooks as not applicable for a hookless triage
 adapter (#752)

opencode-http observes over SSE and never writes an event channel, so the
non-completed-session diagnostic reported `hook events: none` for every
session it ran, pointing the operator at a relay it does not use. Gate the
event scan on the triage adapter's observation mechanism.
---
 CHANGELOG.md           |  3 ++-
 src/bmad_loop/sweep.py | 11 ++++++++++-
 tests/test_sweep.py    | 18 ++++++++++++++++++
 3 files changed, 30 insertions(+), 2 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 927758722..456513222 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -67,7 +67,8 @@ breaking changes may land in a minor release.
 
 - 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; routing is unchanged (#752).
+  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.
 
diff --git a/src/bmad_loop/sweep.py b/src/bmad_loop/sweep.py
index f6970c9cb..9bbc4a33e 100644
--- a/src/bmad_loop/sweep.py
+++ b/src/bmad_loop/sweep.py
@@ -3919,7 +3919,9 @@ def _session_failure_diagnostic(
         document read-only.
 
         ``hook_events``: ``none`` | ``session-start-without-stop`` | ``stop`` |
-        ``no-session-start-or-stop`` | ``unreadable: <reason>``, counted over both
+        ``no-session-start-or-stop`` | ``unreadable: <reason>`` |
+        ``n/a (<observation> observation)`` for a triage adapter that does not
+        observe through hooks and so writes no events to count. Counted over both
         the out-of-tree channel and the legacy in-tree ``<run_dir>/events`` through
         ``signals.is_session_event`` — this attempt's session id and launch floor,
         so an earlier healthy attempt's events cannot mask this failure.
@@ -3955,6 +3957,13 @@ def _session_failure_diagnostic(
                         diagnostic["artifact_error"] = _bounded("; ".join(errors))
                     else:
                         diagnostic["artifact"] = "valid"
+        # An adapter that does not observe through hooks (opencode-http's SSE)
+        # never writes either event channel, so an empty scan is no evidence about
+        # it; `none` would send its operator debugging a relay it does not use.
+        observation = self.adapters["triage"].observation
+        if observation != "hook-signal":
+            diagnostic["hook_events"] = f"n/a ({observation or 'unknown'} observation)"
+            return diagnostic
         try:
             events = session_events(
                 events_dir_for(self.paths.project, self.run_dir.name),
diff --git a/tests/test_sweep.py b/tests/test_sweep.py
index 881ca2536..136f75395 100644
--- a/tests/test_sweep.py
+++ b/tests/test_sweep.py
@@ -5430,6 +5430,24 @@ def test_non_completed_triage_reports_hook_evidence_distinctly(project, primary,
     assert diag["hook_event_count"] == len(primary)
 
 
+def test_non_completed_triage_on_a_hookless_adapter_reports_hooks_not_applicable(project):
+    """opencode-http observes over SSE and never writes an event channel, so an
+    empty scan says nothing about its session: the diagnostic reports hooks as not
+    applicable instead of `none`, which would blame a relay it does not use.
+
+    Ablation guard: drop the observation gate and this fails on `hook_events`."""
+    write_ledger(project, {"DW-1": "open"})
+    engine, adapter = make_sweep(project, [failed_session_effect()] * 2)
+    adapter.observation = "sse"
+    engine.run()
+
+    diag = _decisions(engine, "triage-decision")[-1]["diagnostic"]
+    assert diag["hook_events"] == "n/a (sse observation)"
+    assert "hook_event_count" not in diag
+    assert diag["artifact"] == "missing"
+    assert "hook events: n/a (sse observation)]" in engine.state.paused_reason
+
+
 def test_failed_triage_after_a_healthy_attempt_reports_only_its_own_evidence(project):
     """Attempt 1 ran cleanly — SessionStart, Stop, a result.json on disk — but its
     plan failed validation; attempt 2 then timed out having left nothing. The