diff --git a/.github/scripts/run-baseline-e2e.sh b/.github/scripts/run-baseline-e2e.sh index b99735925..4e9767fdb 100755 --- a/.github/scripts/run-baseline-e2e.sh +++ b/.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 diff --git a/.github/scripts/watch-progress.py b/.github/scripts/watch-progress.py new file mode 100644 index 000000000..6420e7764 --- /dev/null +++ b/.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()) diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml index 436c31dfe..39defe397 100644 --- a/.github/workflows/e2e.yml +++ b/.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) diff --git a/tests/e2e/baseline/conftest.py b/tests/e2e/baseline/conftest.py index 1feb454b0..baa4d37d9 100644 --- a/tests/e2e/baseline/conftest.py +++ b/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. diff --git a/tests/e2e/baseline/smoke/matrix/test_rehydration_partial.py b/tests/e2e/baseline/smoke/matrix/test_rehydration_partial.py index 153fcfe80..346971ed3 100644 --- a/tests/e2e/baseline/smoke/matrix/test_rehydration_partial.py +++ b/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, ) diff --git a/tests/framework_conformance/test_e2e_ci_scripts.py b/tests/framework_conformance/test_e2e_ci_scripts.py index 052457217..d872cd24e 100644 --- a/tests/framework_conformance/test_e2e_ci_scripts.py +++ b/tests/framework_conformance/test_e2e_ci_scripts.py @@ -24,6 +24,8 @@ _EMIT_LANE_MATRIX = CI_SCRIPTS / "emit-lane-matrix.py" _READ_MENTIONS = CI_SCRIPTS / "read-integrations-mentions.sh" +_RUN_BASELINE_E2E = CI_SCRIPTS / "run-baseline-e2e.sh" +_WATCH_PROGRESS = CI_SCRIPTS / "watch-progress.py" _ROSTER = Path(".github") / "integrations-team.txt" # POSIX-shell only. On Windows, `shutil.which("bash")` finds System32\bash.exe — @@ -148,3 +150,96 @@ def test_mentions_reader_emits_at_handles_for_a_real_roster(tmp_path: Path) -> N assert result.returncode == 0 assert (tmp_path / "out.txt").read_text().strip() == "mentions=@alice @bob" + + +def _watch_progress( + tmp_path: Path, code: str, *, idle_seconds: float +) -> subprocess.CompletedProcess[str]: + """Run the shipped progress watchdog around a tiny deterministic child.""" + return subprocess.run( + [ + sys.executable, + str(_WATCH_PROGRESS), + "--idle-seconds", + str(idle_seconds), + "--diagnostic", + str(tmp_path / "diagnostic.json"), + "--", + sys.executable, + "-c", + code, + ], + cwd=REPO_ROOT, + capture_output=True, + text=True, + ) + + +def test_progress_watchdog_terminates_a_silent_child_with_safe_diagnostic( + tmp_path: Path, +) -> None: + """A harness hang must be red and identify only the safe current node id.""" + result = _watch_progress( + tmp_path, + "import time; print('E2E_PROGRESS nodeid=tests/e2e/test_hang.py::test_hang', flush=True); time.sleep(2)", + idle_seconds=0.1, + ) + + assert result.returncode == 124 + assert "current node: tests/e2e/test_hang.py::test_hang" in result.stderr + diagnostic = (tmp_path / "diagnostic.json").read_text(encoding="utf-8") + assert '"kind": "e2e_progress_timeout"' in diagnostic + assert '"nodeid": "tests/e2e/test_hang.py::test_hang"' in diagnostic + + +def test_progress_watchdog_preserves_a_completed_child_output(tmp_path: Path) -> None: + """The watchdog observes progress; it does not change a healthy command's verdict.""" + result = _watch_progress( + tmp_path, + "print('E2E_PROGRESS nodeid=tests/e2e/test_ok.py::test_ok', flush=True)", + idle_seconds=1, + ) + + assert result.returncode == 0 + assert "E2E_PROGRESS nodeid=tests/e2e/test_ok.py::test_ok" in result.stdout + assert not (tmp_path / "diagnostic.json").exists() + + +@posix_shell_only +def test_baseline_runner_keeps_a_progress_timeout_red_without_scorecard( + tmp_path: Path, +) -> None: + """A killed pytest child cannot become a retry pass or a green empty fragment.""" + scripts = tmp_path / ".github" / "scripts" + scripts.mkdir(parents=True) + shutil.copy(_RUN_BASELINE_E2E, scripts) + shutil.copy(_WATCH_PROGRESS, scripts) + fake_bin = tmp_path / "bin" + fake_bin.mkdir() + fake_uv = fake_bin / "uv" + fake_uv.write_text("#!/usr/bin/env bash\nsleep 2\n", encoding="utf-8") + fake_uv.chmod(0o755) + attempts = tmp_path / "artifacts" / "attempts" + diagnostics = tmp_path / "artifacts" / "diagnostics" + result = subprocess.run( + ["bash", str(scripts / _RUN_BASELINE_E2E.name)], + cwd=tmp_path, + env={ + **os.environ, + "PATH": os.pathsep.join( + [str(fake_bin), str(Path(sys.executable).parent), os.environ["PATH"]] + ), + "ATTEMPT1": str(attempts / "one.json"), + "ATTEMPT2": str(attempts / "two.json"), + "FINAL": str(tmp_path / "artifacts" / "scorecard-core-ubuntu.json"), + "PROGRESS_DIAGNOSTIC1": str(diagnostics / "one.json"), + "PROGRESS_DIAGNOSTIC2": str(diagnostics / "two.json"), + "E2E_PROGRESS_DEADLINE_SECONDS": "0.1", + }, + capture_output=True, + text=True, + ) + + assert result.returncode == 124 + assert (diagnostics / "one.json").exists() + assert not (tmp_path / "artifacts" / "scorecard-core-ubuntu.json").exists()