orchestrator/agent-team/tests/test_runaway_monitor.py

"""Unit tests for agent_team.draft_pr_monitor (P3 box-integration, A4).

Covers the draft-PR runaway/stale sweep: it ALARMs when > 3 draft PRs open
within 15 minutes, reminds on a draft PR idle > 7 days, applies flapping backoff
(one ALARM per cooldown, one reminder per PR per cooldown), and never auto-closes
or self-stops. No network: every side effect (alarm, reminder) is an injected
callable, exactly as the module's contract promises.
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone

from agent_team.draft_pr_monitor import (
    DEFAULT_ALARM_COOLDOWN,
    DEFAULT_STALE_COOLDOWN,
    DraftPr,
    MonitorAction,
    MonitorMemory,
    run_draft_pr_monitor,
)

# A fixed "now"; ISO strings mirror the GitHub API / ledger (UTC).
_NOW = datetime(2026, 6, 23, 12, 0, 0, tzinfo=timezone.utc)


def _ago(**kwargs: float) -> str:
    return (_NOW - timedelta(**kwargs)).isoformat()


class _Recorder:
    """Records the runaway counts / stale PRs handed to the side-effect hooks."""

    def __init__(self) -> None:
        self.alarms: list[int] = []
        self.reminders: list[int] = []

    def on_alarm(self, opened_in_window: int) -> None:
        self.alarms.append(opened_in_window)

    def on_stale_reminder(self, pr: DraftPr) -> None:
        self.reminders.append(pr.number)


def _open_burst(count: int, *, within_minutes: int = 5) -> list[DraftPr]:
    """``count`` draft PRs all opened ``within_minutes`` ago (inside the window)."""
    return [
        DraftPr(
            number=i,
            opened_at=_ago(minutes=within_minutes),
            updated_at=_NOW.isoformat(),
        )
        for i in range(count)
    ]


# --------------------------------------------------------------------------- #
# Runaway ALARM threshold (> 3 within 15 min)
# --------------------------------------------------------------------------- #


def test_runaway_alarm_fires_above_threshold() -> None:
    rec = _Recorder()
    report = run_draft_pr_monitor(
        _open_burst(4),
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.alarmed == 1
    assert rec.alarms == [4]
    assert report.outcomes[0].action is MonitorAction.ALARMED


def test_exactly_threshold_does_not_alarm() -> None:
    """The rule is STRICTLY greater than 3 — exactly 3 must NOT alarm."""
    rec = _Recorder()
    report = run_draft_pr_monitor(
        _open_burst(3),
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.alarmed == 0
    assert rec.alarms == []


def test_prs_opened_outside_window_are_not_counted() -> None:
    """5 PRs but only 2 inside the 15-min window -> no alarm."""
    rec = _Recorder()
    prs = [
        DraftPr(number=1, opened_at=_ago(minutes=2)),
        DraftPr(number=2, opened_at=_ago(minutes=10)),
        DraftPr(number=3, opened_at=_ago(minutes=20)),
        DraftPr(number=4, opened_at=_ago(minutes=40)),
        DraftPr(number=5, opened_at=_ago(hours=3)),
    ]
    report = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.opened_in_window == 2
    assert report.alarmed == 0
    assert rec.alarms == []


def test_unparseable_opened_at_is_skipped_for_runaway() -> None:
    """A PR with a junk/missing opened_at is not counted and does not crash."""
    rec = _Recorder()
    prs = _open_burst(4) + [DraftPr(number=99, opened_at="not-a-date")]
    report = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    # The 4 valid PRs still breach; the junk one is just ignored for the count.
    assert report.opened_in_window == 4
    assert report.alarmed == 1


# --------------------------------------------------------------------------- #
# Flapping backoff — ALARM at most once per cooldown
# --------------------------------------------------------------------------- #


def test_runaway_alarm_suppressed_within_cooldown() -> None:
    rec = _Recorder()
    mem = MonitorMemory()
    prs = _open_burst(5)

    first = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW,
    )
    assert first.alarmed == 1

    # A second tick a minute later, condition still breaching -> suppressed.
    second = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW + timedelta(minutes=1),
    )
    assert second.alarmed == 0
    assert second.alarm_suppressed == 1
    assert rec.alarms == [5]  # only the first post


def test_runaway_alarm_refires_after_cooldown() -> None:
    rec = _Recorder()
    mem = MonitorMemory()
    prs = _open_burst(5, within_minutes=1)

    run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW,
    )
    # Past the cooldown, a still-breaching condition ALARMs again. Re-anchor the
    # PRs so they remain inside the window relative to the later "now".
    later = _NOW + DEFAULT_ALARM_COOLDOWN + timedelta(seconds=1)
    prs_later = [
        DraftPr(number=p.number, opened_at=(later - timedelta(minutes=1)).isoformat())
        for p in prs
    ]
    report = run_draft_pr_monitor(
        prs_later,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=later,
    )
    assert report.alarmed == 1
    assert len(rec.alarms) == 2


# --------------------------------------------------------------------------- #
# Stale reminder (> 7 days idle), never auto-closes
# --------------------------------------------------------------------------- #


def test_stale_pr_reminds_once() -> None:
    rec = _Recorder()
    prs = [DraftPr(number=7, opened_at=_ago(days=30), updated_at=_ago(days=8))]
    report = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.stale_reminded == 1
    assert rec.reminders == [7]


def test_fresh_pr_does_not_remind() -> None:
    rec = _Recorder()
    prs = [DraftPr(number=7, opened_at=_ago(days=30), updated_at=_ago(days=6))]
    report = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.stale_reminded == 0
    assert rec.reminders == []


def test_exactly_7_days_does_not_remind() -> None:
    """The rule is strictly MORE than 7 days idle."""
    rec = _Recorder()
    prs = [DraftPr(number=7, updated_at=_ago(days=7))]
    report = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.stale_reminded == 0


def test_stale_reminder_suppressed_within_cooldown() -> None:
    rec = _Recorder()
    mem = MonitorMemory()
    prs = [DraftPr(number=7, updated_at=_ago(days=10))]

    run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW,
    )
    second = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW + timedelta(hours=1),
    )
    assert second.stale_reminded == 0
    assert second.stale_suppressed == 1
    assert rec.reminders == [7]  # only the first


def test_stale_reminder_refires_after_cooldown() -> None:
    rec = _Recorder()
    mem = MonitorMemory()

    run_draft_pr_monitor(
        [DraftPr(number=7, updated_at=_ago(days=10))],
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW,
    )
    later = _NOW + DEFAULT_STALE_COOLDOWN + timedelta(seconds=1)
    report = run_draft_pr_monitor(
        [DraftPr(number=7, updated_at=(later - timedelta(days=10)).isoformat())],
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=later,
    )
    assert report.stale_reminded == 1
    assert len(rec.reminders) == 2


def test_unparseable_updated_at_is_skipped_for_stale() -> None:
    rec = _Recorder()
    prs = [DraftPr(number=7, updated_at="garbage")]
    report = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.stale_reminded == 0
    assert rec.reminders == []


# --------------------------------------------------------------------------- #
# Side-effect isolation + watermark-not-advanced-on-error
# --------------------------------------------------------------------------- #


def test_alarm_side_effect_error_is_isolated_and_retried() -> None:
    """A raising on_alarm yields ERRORED and does NOT advance the watermark."""
    calls: list[int] = []

    def boom(opened_in_window: int) -> None:
        calls.append(opened_in_window)
        raise RuntimeError("slack down")

    mem = MonitorMemory()
    prs = _open_burst(5)
    report = run_draft_pr_monitor(
        prs,
        on_alarm=boom,
        on_stale_reminder=lambda pr: None,
        memory=mem,
        now=_NOW,
    )
    assert report.errored == 1
    assert report.alarmed == 0
    # Watermark not advanced -> a retry the very next tick (no cooldown lock-in).
    assert mem.last_alarm_at is None
    report2 = run_draft_pr_monitor(
        prs,
        on_alarm=boom,
        on_stale_reminder=lambda pr: None,
        memory=mem,
        now=_NOW + timedelta(seconds=30),
    )
    assert report2.errored == 1
    assert len(calls) == 2


def test_stale_side_effect_error_does_not_abort_sweep() -> None:
    """A reminder that raises for one PR does not stop the runaway ALARM."""
    rec = _Recorder()

    def boom(pr: DraftPr) -> None:
        raise RuntimeError("slack down")

    prs = _open_burst(5) + [DraftPr(number=50, updated_at=_ago(days=10))]
    report = run_draft_pr_monitor(
        prs,
        on_alarm=rec.on_alarm,
        on_stale_reminder=boom,
        now=_NOW,
    )
    # Runaway still alarms; the stale reminder errored but was isolated.
    assert report.alarmed == 1
    assert report.errored == 1


# --------------------------------------------------------------------------- #
# Never auto-closes / self-stops; memory hygiene
# --------------------------------------------------------------------------- #


def test_monitor_takes_no_infrastructure_action() -> None:
    """The only effects are the two injected notify callables — nothing else.

    There is no close/stop/dispatch seam on the monitor; this asserts the sweep
    exposes ONLY the alarm + reminder hooks (a regression guard against adding a
    self-acting side effect).
    """
    import inspect

    sig = inspect.signature(run_draft_pr_monitor)
    effect_params = {p for p in sig.parameters if p.startswith("on_")}
    assert effect_params == {"on_alarm", "on_stale_reminder"}


def test_memory_prunes_gone_prs() -> None:
    """A PR no longer present has its backoff watermark pruned (bounded growth)."""
    rec = _Recorder()
    mem = MonitorMemory()

    run_draft_pr_monitor(
        [DraftPr(number=7, updated_at=_ago(days=10))],
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW,
    )
    assert 7 in mem.last_reminded_at

    # Next pass: PR #7 is gone (merged/closed by a human). Its watermark prunes.
    run_draft_pr_monitor(
        [],
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        memory=mem,
        now=_NOW + timedelta(minutes=1),
    )
    assert 7 not in mem.last_reminded_at


def test_no_draft_prs_is_quiet() -> None:
    rec = _Recorder()
    report = run_draft_pr_monitor(
        [],
        on_alarm=rec.on_alarm,
        on_stale_reminder=rec.on_stale_reminder,
        now=_NOW,
    )
    assert report.outcomes == []
    assert rec.alarms == []
    assert rec.reminders == []