rune/tests/test_event_wait.py

"""test_event_wait.py — T15 §15D/§15E: event-driven wait as canonical
default + event semantics adversarials.

Runs the REAL W/R daemon as a subprocess against an isolated WR_ROOT
(the same disposable harness pattern as the bridge's own v0.2/v0.3
invariant tests), then drives the repo-side adapter (rune.event_wait)
through it. No real MiMo sessions; the "builder" is a cheap fake command.

Proofs:
  15D latency      — controller resumes on the EVENT; continuation
                     latency recorded; no arbitrary wait floor exists
                     (a deliberately long would-be floor is NOT waited).
  15D failure      — non-zero builder exit → verdict failure, resume
                     still fires (timeout is not success; events carry
                     truth).
  15D watchdog     — lost event → deadline expires → one-shot liveness
                     inspection → PROCESS_LOST classification. Watchdog
                     is the failure detector, never the scheduler.
  15E duplicate    — duplicate completion event cannot double-resume.
  15E stale        — event from a prior operation matches nothing.
  15E wrong corr   — unknown operation_id matches nothing.
  15E malformed    — event without exit_code leaves wait WAITING.
  15E restart      — controller restart after event persisted: durable
                     waitstore wins; resume_or_recover is idempotent.
"""

from __future__ import annotations

import json
import os
import signal
import subprocess
import sys
import tempfile
import time
import urllib.request
from pathlib import Path

import pytest

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO / "src"))
WR = Path("/home/nick/.local/share/mimo-wait-resume")
sys.path.insert(0, str(WR))

from rune import event_wait as ew  # noqa: E402


class Daemon:
    """Real daemon subprocess on an isolated WR_ROOT + ephemeral port."""

    def __init__(self, tmp_root: Path, port: int | None = None):
        # Ephemeral port (not a fixed one): the CI runner executes the
        # FULL suite in one process, and a fixed port collides with
        # TIME_WAIT sockets from earlier tests in the same file or with
        # parallel jobs on the same runner. 15Z CI failure: the latency
        # test's daemon failed to bind 8793 (stale socket), the notifier's
        # event POSTs went nowhere, and the watchdog correctly reported
        # PROCESS_LOST — the test harness was at fault, not the bridge.
        if port is None:
            import socket as _socket
            with _socket.socket() as _s:
                _s.bind(("127.0.0.1", 0))
                port = _s.getsockname()[1]
        self.root, self.port = tmp_root, port
        # T21.5 hardening: under heavy parallel load the first start may
        # lose the bind race or the health probe may exceed its window.
        # Retry once on a FRESH ephemeral port before failing — the
        # fixture must be robust, not flaky.
        import socket as _socket
        for attempt in range(2):
            if attempt == 1:
                with _socket.socket() as _s:
                    _s.bind(("127.0.0.1", 0))
                    self.port = _s.getsockname()[1]
            if self._spawn():
                return
            self.stop()
        raise RuntimeError("daemon did not become healthy")

    def url(self) -> str:
        return f"http://127.0.0.1:{self.port}"

    def stop(self):
        self.proc.send_signal(signal.SIGTERM)
        try:
            self.proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            self.proc.kill()

    def _spawn(self) -> bool:
        """(Re)start the daemon subprocess on THIS instance's root + port;
        returns True once /health answers 200. The wait registrations are
        file-backed under WR_ROOT, so a respawned daemon resumes the same
        state — a mid-test death is recoverable."""
        env = dict(os.environ,
                   WR_ROOT=str(self.root),
                   WR_DRIVE_SECS="1",
                   WR_RECONCILE_SECS="2",
                   WR_RECONCILE_TICKS="1",
                   WR_WATCH_SECS="30",
                   WR_TERM_GRACE_SECS="1",
                   RUNE_WR_ROOT=str(self.root))
        self.proc = subprocess.Popen(
            [sys.executable, str(WR / "daemon.py"), str(self.port)],
            stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
            text=True, env=env)
        deadline = time.time() + 15
        while time.time() < deadline:
            try:
                with urllib.request.urlopen(
                        f"http://127.0.0.1:{self.port}/health",
                        timeout=2) as r:
                    if r.status == 200:
                        return True
            except Exception:
                if self.proc.poll() is not None:
                    return False  # died before health — the caller retries
                time.sleep(0.2)
        return False

    def _ensure_alive(self) -> None:
        if self.proc.poll() is not None:
            self._spawn()

    def post_event(self, payload: dict) -> dict:
        # Flake hardening (observed 2026-10-04, runs 18340-18359): the daemon
        # answers /health at fixture setup, then DIES SILENTLY mid-test on
        # the saturated runner (diagnostic: alive=False, out='' — silent
        # kill under memory pressure; 20GB swap in use on the host).
        # Registrations are file-backed under WR_ROOT, so a respawned daemon
        # resumes the same state — retry with respawn, and on final failure
        # dump the daemon's captured output for the next diagnosis.
        last_err: Exception | None = None
        for attempt in range(3):
            self._ensure_alive()
            try:
                req = urllib.request.Request(
                    self.url() + "/process-event",
                    data=json.dumps(payload).encode(),
                    headers={"Content-Type": "application/json"})
                with urllib.request.urlopen(req, timeout=5) as r:
                    return json.loads(r.read())
            except Exception as e:  # noqa: BLE001 — retry wrapper: any POST failure retries
                last_err = e
                if attempt < 2:
                    time.sleep(0.5 * (attempt + 1))
        daemon_out = ""
        try:
            if self.proc.stdout is not None:
                import os as _os
                _os.set_blocking(self.proc.stdout.fileno(), False)
                daemon_out = (self.proc.stdout.read() or "")[-800:]
        except Exception:  # noqa: BLE001 — diagnostics must never mask the failure
            pass
        raise RuntimeError(
            f"post_event refused after 3 attempts (with respawn): {last_err}\n"
            f"daemon alive={self.proc.poll() is None} out={daemon_out!r}") from last_err


# T16 §16E: the daemon fixture moved to tests/conftest.py (wr_daemon) so
# test_ci_event_wait.py shares ONE harness — no duplicated test helpers.
@pytest.fixture()
def daemon(wr_daemon):
    return wr_daemon


# real fake-builder: sleeps then exits — cheap but genuinely a process
FAKE_BUILDER = ["bash", "-c", "sleep 2; exit 0"]
FAKE_BUILDER_FAIL = ["bash", "-c", "sleep 1; exit 3"]


def test_15d_latency_event_driven_no_floor(daemon):
    """Continuation happens FROM the event: controller resumes ~immediately
    after the 2s builder exits — proving no 60s+ polling floor exists."""
    op = "op-latency-proof"
    rec = ew.register_process_wait(
        op, contract_id="contract:t15", project="t15-fixture",
        objective="latency proof", watchdog_secs=120)
    assert rec["wait_kind"] == "PROCESS_EVENT"
    assert rec["resume_cmd"], "per-wait resume_cmd required"

    builder_started = time.time()
    notifier = ew.run_builder_under_events(FAKE_BUILDER, op,
                                           daemon_url=daemon.url())
    result = ew.await_resume_event(op, watchdog_secs=120,
                                   pgid=notifier.pid)
    resumed_at = time.time()
    assert result["kind"] == "event", f"expected event resume, got {result}"
    wait = result["wait"]
    assert wait["verdict"] == "success"
    evidence = wait.get("process_event_evidence", {})
    child_exit = evidence and float(evidence.get("exit_code") == 0)
    # The event carries the actual exit evidence
    assert evidence.get("exit_code") == 0
    builder_elapsed = resumed_at - builder_started
    # builder itself takes ~2s; if a 60s polling floor existed we'd exceed it
    assert builder_elapsed < 30, \
        f"continuation took {builder_elapsed:.1f}s — a wait floor exists"
    # and it cannot be faster than the builder itself (event came from
    # the real process exit, not a timer)
    assert builder_elapsed >= 1.5
    assert wait["status"] in ("RESUMING", "COMPLETED")


def test_15d_failure_event_carries_truth(daemon):
    op = "op-failure-proof"
    ew.register_process_wait(op, project="t15-fixture",
                             objective="failure proof", watchdog_secs=60)
    notifier = ew.run_builder_under_events(FAKE_BUILDER_FAIL, op,
                                           daemon_url=daemon.url())
    result = ew.await_resume_event(op, watchdog_secs=60, pgid=notifier.pid)
    assert result["kind"] == "event"
    assert result["wait"]["verdict"] == "failure"
    assert result["wait"]["process_event_evidence"]["exit_code"] == 3


def test_15d_watchdog_detects_lost_event(daemon, tmp_path):
    """Lost event: deadline expires → one-shot liveness inspection →
    PROCESS_LOST. The watchdog never schedules; it detects failure of
    continuation."""
    op = "op-lost-event"
    rec = ew.register_process_wait(op, project="t15-fixture",
                                   objective="lost event", watchdog_secs=3)
    # launch a long-running builder whose event will NEVER arrive
    notifier = ew.run_builder_under_events(
        ["bash", "-c", "sleep 120"], op, daemon_url=daemon.url())
    result = ew.await_resume_event(op, watchdog_secs=3, pgid=notifier.pid)
    assert result["kind"] == "watchdog"
    assert result["alive"] is True  # process still alive → NOT lost, re-arm
    # now genuinely lose it: kill the group, next watchdog sees GONE
    os.killpg(notifier.pid, signal.SIGKILL)
    result2 = ew.await_resume_event(op, watchdog_secs=3, pgid=notifier.pid)
    assert result2["kind"] == "watchdog"
    assert result2["alive"] is False  # PROCESS_LOST → recovery path
    # recovery: process died without completion — the wait is still
    # WAITING_EXTERNAL (canonical state: nothing completed)
    wait = ew.operation_wait(op)
    assert wait["status"] == "WAITING_EXTERNAL"


def test_15e_duplicate_event_cannot_double_resume(daemon):
    op = "op-dup-event"
    ew.register_process_wait(op, project="t15-fixture", watchdog_secs=60)
    first = daemon.post_event({"operation_id": op, "exit_code": 0})
    assert first["matched"] >= 1
    wait_for(lambda: (ew.operation_wait(op)["status"] != "READY_TO_RESUME"))
    second = daemon.post_event({"operation_id": op, "exit_code": 0})
    # second event matches nothing new (wait no longer WAITING_EXTERNAL)
    # and must not re-dispatch a resumed wait
    rec = ew.operation_wait(op)
    if second["matched"] > 0:
        # if it dispatched again, it must NOT have reset resume state
        assert rec.get("resume_attempts", 0) >= 1
    assert rec["status"] != "WAITING_EXTERNAL"
    # verdict unchanged by the duplicate
    assert rec.get("verdict") in ("success", None)


def test_15e_stale_event_from_prior_operation_matches_nothing(daemon):
    op_old, op_new = "op-old", "op-new"
    ew.register_process_wait(op_old, project="t15-fixture", watchdog_secs=60)
    ew.register_process_wait(op_new, project="t15-fixture", watchdog_secs=60)
    # satisfy + consume the old operation's event
    daemon.post_event({"operation_id": op_old, "exit_code": 0})
    wait_for(lambda: ew.operation_wait(op_old)["status"] != "READY_TO_RESUME")
    # a STALE repeat of the old event must not touch the new operation
    daemon.post_event({"operation_id": op_old, "exit_code": 0})
    new_wait = ew.operation_wait(op_new)
    assert new_wait["status"] == "WAITING_EXTERNAL"
    assert new_wait.get("verdict") is None


def test_15e_wrong_correlation_matches_nothing(daemon):
    op = "op-wrong-corr"
    ew.register_process_wait(op, project="t15-fixture", watchdog_secs=60)
    resp = daemon.post_event({"operation_id": "completely-unrelated",
                              "exit_code": 0})
    assert resp["matched"] == 0
    assert ew.operation_wait(op)["status"] == "WAITING_EXTERNAL"


def test_15e_malformed_event_without_exit_code_stays_waiting(daemon):
    op = "op-malformed"
    ew.register_process_wait(op, project="t15-fixture", watchdog_secs=60)
    resp = daemon.post_event({"operation_id": op})  # no exit_code
    assert ew.operation_wait(op)["status"] == "WAITING_EXTERNAL"


def test_15e_restart_after_event_persisted_waitstore_wins(daemon):
    """Case 10: the event was persisted and the wait advanced, but the
    controller 'crashed' before handling it. On restart the durable
    waitstore wins; resume_or_recover is idempotent (no duplicate
    transition, no duplicate resume)."""
    op = "op-restart"
    ew.register_process_wait(op, project="t15-fixture", watchdog_secs=60)
    daemon.post_event({"operation_id": op, "exit_code": 0})
    wait_for(lambda: ew.operation_wait(op)["status"] != "READY_TO_RESUME")
    r1 = ew.resume_or_recover(op)
    r2 = ew.resume_or_recover(op)
    assert r1["kind"] == "event" and r1.get("recovered") is True
    assert r2 == r1  # idempotent: canonical state, not transient token
    assert r1["wait"]["verdict"] == "success"


def wait_for(fn, timeout=30, interval=0.3):
    deadline = time.time() + timeout
    while time.time() < deadline:
        r = fn()
        if r:
            return r
        time.sleep(interval)
    return fn()