fix: make baseline E2E hangs fail safely

.github/scripts/run-baseline-e2e.sh

@@ -18,9 +18,26 @@ set -uo pipefail
 : "${ATTEMPT1:?ATTEMPT1 scorecard path is required}"
 : "${ATTEMPT2:?ATTEMPT2 scorecard path is required}"
 : "${FINAL:?FINAL scorecard path is required}"
+: "${PROGRESS_DIAGNOSTIC1:?first-attempt progress diagnostic path is required}"
+: "${PROGRESS_DIAGNOSTIC2:?retry progress diagnostic path is required}"
+
+# The workflow supplies this one conservative ceiling. A per-test timeout remains
+# the primary contract; this only catches a harness that stops enforcing it.
+E2E_PROGRESS_DEADLINE_SECONDS="${E2E_PROGRESS_DEADLINE_SECONDS:-1800}"
 
 mkdir -p artifacts/attempts
-BAND_E2E_SCORECARD_JSON="$ATTEMPT1" uv run pytest tests/e2e/baseline/ -v -s --no-cov
+run_pytest() {
+  local scorecard="$1"
+  local diagnostic="$2"
+  shift 2
+  BAND_E2E_SCORECARD_JSON="$scorecard" \
+    python .github/scripts/watch-progress.py \
+      --idle-seconds "$E2E_PROGRESS_DEADLINE_SECONDS" \
+      --diagnostic "$diagnostic" \
+      -- uv run pytest tests/e2e/baseline/ -v -s --no-cov "$@"
+}
+
+run_pytest "$ATTEMPT1" "$PROGRESS_DIAGNOSTIC1"
 code=$?
 if [ "$code" -ne 0 ]; then
   # A retry only helps for one-off flakiness (a rate-limit window, a cold start).
@@ -39,8 +56,7 @@ if [ "$code" -ne 0 ]; then
     # the retry into a second full live lane -- double the provider spend and wall
     # clock, against a leg that carries a wall-clock cap. Deselecting instead is the
     # honest reading: a retry with nothing identifiable to retry has nothing to add.
-    BAND_E2E_SCORECARD_JSON="$ATTEMPT2" uv run pytest tests/e2e/baseline/ -v -s --no-cov \
-      --last-failed --lfnf=none
+    run_pytest "$ATTEMPT2" "$PROGRESS_DIAGNOSTIC2" --last-failed --lfnf=none
     retry_code=$?
     # Exit 5 is pytest's "no tests collected" -- here, --lfnf=none deselecting
     # everything (the no-cache case above). That says nothing about the lane, so

.github/scripts/watch-progress.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+"""Run a command and fail it if its safe progress signal stops advancing."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import signal
+import subprocess
+import sys
+import time
+from pathlib import Path
+from queue import Empty, Queue
+from threading import Thread
+from typing import TextIO
+
+
+_PROGRESS_PREFIX = "E2E_PROGRESS nodeid="
+_WATCHDOG_EXIT_CODE = 124
+
+
+def _arguments() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--idle-seconds", type=float, required=True)
+    parser.add_argument("--diagnostic", type=Path, required=True)
+    parser.add_argument("command", nargs=argparse.REMAINDER)
+    args = parser.parse_args()
+    if args.idle_seconds <= 0:
+        parser.error("--idle-seconds must be positive")
+    if not args.command or args.command[0] != "--" or len(args.command) == 1:
+        parser.error("a command must follow --")
+    args.command = args.command[1:]
+    return args
+
+
+def _read_lines(stream: TextIO, lines: Queue[str | None]) -> None:
+    for line in iter(stream.readline, ""):
+        lines.put(line)
+    lines.put(None)
+
+
+def _terminate_tree(process: subprocess.Popen[str]) -> None:
+    if os.name == "nt":
+        subprocess.run(
+            ["taskkill", "/pid", str(process.pid), "/t", "/f"],
+            check=False,
+            capture_output=True,
+        )
+        return
+    try:
+        os.killpg(process.pid, signal.SIGTERM)
+    except ProcessLookupError:
+        return
+    try:
+        process.wait(timeout=10)
+    except subprocess.TimeoutExpired:
+        try:
+            os.killpg(process.pid, signal.SIGKILL)
+        except ProcessLookupError:
+            return
+
+
+def _write_timeout_diagnostic(
+    path: Path, *, nodeid: str | None, elapsed_seconds: float, pid: int
+) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(
+        json.dumps(
+            {
+                "kind": "e2e_progress_timeout",
+                "elapsed_seconds": round(elapsed_seconds, 1),
+                "nodeid": nodeid,
+                "pid": pid,
+            },
+            indent=2,
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+
+
+def main() -> int:
+    args = _arguments()
+    process = subprocess.Popen(
+        args.command,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.STDOUT,
+        text=True,
+        bufsize=1,
+        start_new_session=os.name != "nt",
+        creationflags=(
+            getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) if os.name == "nt" else 0
+        ),
+    )
+    assert process.stdout is not None
+    lines: Queue[str | None] = Queue()
+    Thread(target=_read_lines, args=(process.stdout, lines), daemon=True).start()
+
+    last_progress = time.monotonic()
+    nodeid: str | None = None
+    stream_closed = False
+    while not stream_closed:
+        try:
+            line = lines.get(timeout=min(1.0, args.idle_seconds))
+        except Empty:
+            elapsed = time.monotonic() - last_progress
+            if elapsed < args.idle_seconds:
+                continue
+            _write_timeout_diagnostic(
+                args.diagnostic,
+                nodeid=nodeid,
+                elapsed_seconds=elapsed,
+                pid=process.pid,
+            )
+            sys.stderr.write(
+                "::error::E2E made no progress for "
+                f"{elapsed:.0f}s (current node: {nodeid or 'unknown'}); terminating it\n"
+            )
+            sys.stderr.flush()
+            _terminate_tree(process)
+            return _WATCHDOG_EXIT_CODE
+        if line is None:
+            stream_closed = True
+            continue
+        sys.stdout.write(line)
+        sys.stdout.flush()
+        last_progress = time.monotonic()
+        if line.startswith(_PROGRESS_PREFIX):
+            nodeid = line.removeprefix(_PROGRESS_PREFIX).strip()
+
+    return process.wait()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

.github/workflows/e2e.yml

@@ -190,6 +190,9 @@ jobs:
       # @pytest.mark.timeout(extra=...). See tests/e2e/baseline/conftest.py.
       # The registry scopes which adapters run in this lane.
       BAND_E2E_LANE: ${{ matrix.lane }}
+      # Per-test timeouts are the primary boundary. This larger no-progress
+      # watchdog catches a stuck pytest process that failed to enforce one.
+      E2E_PROGRESS_DEADLINE_SECONDS: "1800"
       # The dev-crewai and dev-parlant venvs each lack the other frameworks;
       # tolerate missing framework configs there instead of failing fast (mirrors
       # ci.yml's crewai and parlant jobs).
@@ -283,6 +286,8 @@ jobs:
         ATTEMPT1: artifacts/attempts/scorecard-${{ matrix.lane }}-${{ matrix.os }}-1.json
         ATTEMPT2: artifacts/attempts/scorecard-${{ matrix.lane }}-${{ matrix.os }}-2.json
         FINAL: artifacts/scorecard-${{ matrix.lane }}-${{ matrix.os }}.json
+        PROGRESS_DIAGNOSTIC1: artifacts/diagnostics/progress-${{ matrix.lane }}-${{ matrix.os }}-1.json
+        PROGRESS_DIAGNOSTIC2: artifacts/diagnostics/progress-${{ matrix.lane }}-${{ matrix.os }}-2.json
       run: bash .github/scripts/run-baseline-e2e.sh
 
     # The backends lane also exercises the editor-facing ACP path (codex-acp).
@@ -336,6 +341,7 @@ jobs:
         path: |
           artifacts/scorecard-*.json
           artifacts/environment-*.json
+          artifacts/diagnostics/*.json
         if-no-files-found: ignore
 
   # Fold the per-lane scorecards into one adapter×test grid (pass / fail / skip / N-A)

tests/e2e/baseline/conftest.py

@@ -11,6 +11,7 @@
 from __future__ import annotations
 
 from pathlib import Path
+from typing import Any
 
 import pytest
 
@@ -48,6 +49,9 @@
 from tests.e2e.baseline.settings import BaselineSettings
 from tests.toolkit.timeouts import effective_timeout
 
+
+_terminal_reporter: Any | None = None
+
 # Re-exported fixtures (defined in fixtures/*; imported so pytest registers them).
 __all__ = [
     "adapter_id",
@@ -98,6 +102,12 @@ def pytest_configure(config: pytest.Config) -> None:
     )
 
 
+def pytest_sessionstart(session: pytest.Session) -> None:
+    """Resolve the terminal reporter after pytest has registered its built-ins."""
+    global _terminal_reporter
+    _terminal_reporter = session.config.pluginmanager.get_plugin("terminalreporter")
+
+
 def pytest_runtest_setup(item: pytest.Item) -> None:
     """Gate every baseline test, then resolve any ``@requires(...)`` extras.
 
@@ -121,6 +131,12 @@ def pytest_runtest_setup(item: pytest.Item) -> None:
             require_dep(dep, settings)
 
 
+def pytest_runtest_logstart(nodeid: str, location: tuple[str, int | None, str]) -> None:
+    """Emit a safe current-node marker for CI's no-progress watchdog."""
+    if _terminal_reporter is not None:
+        _terminal_reporter.write_line(f"E2E_PROGRESS nodeid={nodeid}")
+
+
 def pytest_collection_modifyitems(items: list[pytest.Item]) -> None:
     """Guard wiring + schedulability, scope to ``BAND_E2E_LANE``, then apply the
     session event loop + per-turn timeout to every baseline test.

tests/e2e/baseline/smoke/matrix/test_rehydration_partial.py

@@ -7,9 +7,9 @@
 that the single-agent rejoin test cannot exercise:
 
 * Selective/partial rehydration — the rebooted agent (a fresh adapter under the
-  same identity, no in-memory state) still recalls a fact stated before its reboot,
-  which can only have come from the platform rehydrating the room on bootstrap
-  (``/context``). The reboot happens amid a live peer, not in an empty room.
+same identity, no in-memory state) recalls a fact its peer stated *while it was
+offline*. That marker cannot be in the rebooted adapter's persisted backend session,
+so it must have come from platform ``/context`` on bootstrap.
 * Peer continuity — the never-rebooted agent, which stayed up across its neighbour's
   churn, answers a liveness probe unperturbed. Rebooting one participant does not
   disturb the other.
@@ -47,7 +47,9 @@
 scoping); until then two Letta agents cannot keep separate identities in one org.
 
 Wording note: a neutral "note", not a "secret code" (models refuse to echo a
-credential-shaped value — an unrelated false failure).
+credential-shaped value — an unrelated false failure). The peer's setup reply must
+both contain the marker and mention the offline rebooter: platform history is
+agent-scoped, so an unmentioned peer message would not belong in its ``/context``.
 """
 
 from __future__ import annotations
@@ -57,8 +59,6 @@
 
 from tests.e2e.baseline.agents import Adapter, ExcludedAdapter, per_adapter
 from tests.e2e.baseline.smoke.samples.sample_agents import (
-    RECALL,
-    REMEMBER,
     REPLY_PROMPT,
     unique_marker,
 )
@@ -114,7 +114,7 @@ async def test_partial_reboot_preserves_context_and_peer(
     user_ops: UserOps,
     reply_capture: CaptureFactory,
 ) -> None:
-    """Rebooting one agent recalls via rehydration; the peer stays responsive."""
+    """A rebooted agent recalls its offline peer's note; the peer stays responsive."""
     note = unique_marker("note")
 
     # Two distinct identities from the same cell — distinct labels or the generated
@@ -126,29 +126,54 @@ async def test_partial_reboot_preserves_context_and_peer(
         participants=[stayer.id, rebooter.id],
     )
 
+    relay_prompt = (
+        REPLY_PROMPT
+        + f" When asked, send exactly one message that mentions participant "
+        f"'{rebooter.name}' and contains this exact note: {note}."
+    )
+
     # The stayer is UP for the entire test — it never reboots, so it can only answer
     # the liveness probe if a peer's reboot left it undisturbed.
-    async with cell.run_as(stayer):
-        # Rebooter run 1: state the note to the rebooter, then stop it (exit block).
+    async with cell.run_as(stayer, prompt=relay_prompt):
+        # Establish the rebooter's own backend session before it goes offline. The
+        # marker below is created only afterwards, so session resume cannot explain
+        # a successful recall.
         async with cell.run_as(rebooter):
             async with reply_capture(room_id) as capture:
+                mark = capture.messages.snapshot()
                 mid = await user_ops.send_message(
                     room_id,
-                    REMEMBER.format(note=note),
+                    "Please acknowledge that you are ready.",
                     mention_id=rebooter.id,
                     mention_name=rebooter.name,
                 )
-                await capture.wait_for_processed(mid, rebooter.id)
-
-        # Rebooter run 2: a brand-new adapter under the SAME identity — no in-memory
-        # history. A correct recall proves the platform rehydrated the room on
-        # bootstrap, even though the reboot happened alongside a live peer.
+                await capture.wait_for_reply(mid, rebooter.id, since=mark)
+
+        # While the rebooter is offline, the stayer authors a marker-bearing message
+        # that explicitly mentions it. This is the agent-scoped-history precondition:
+        # without the mention, the platform correctly omits the peer message from the
+        # rebooter's /context and a failed recall would be ambiguous.
+        async with reply_capture(room_id) as capture:
+            mark = capture.messages.snapshot()
+            mid = await user_ops.send_message(
+                room_id,
+                f"Please pass the note to {rebooter.name}.",
+                mention_id=stayer.id,
+                mention_name=stayer.name,
+            )
+            replies = await capture.wait_for_reply(mid, stayer.id, since=mark)
+            replies.mentioning(rebooter.id).assert_contains_any([note])
+
+        # Rebooter run 2: a fresh adapter under the SAME identity. Its old backend
+        # session cannot know a marker created while it was offline; a correct recall
+        # therefore proves platform rehydration even for session-resume adapters.
         async with cell.run_as(rebooter):
             async with reply_capture(room_id) as capture:
                 mark = capture.messages.snapshot()  # scope to the recall turn
                 mid = await user_ops.send_message(
                     room_id,
-                    RECALL,
+                    "Earlier, the other participant sent you a short note with a token. "
+                    "Reply with just that token.",
                     mention_id=rebooter.id,
                     mention_name=rebooter.name,
                 )