From e7a2cdfc78a5a88d8aff334c297e53ce3ddcf53a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 1 Jul 2026 08:19:15 +0000 Subject: [PATCH 1/2] Initial plan From a4a9f2d4f6b4ac02d10e7cd2af174253a9aee500 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 1 Jul 2026 08:27:36 +0000 Subject: [PATCH 2/2] =?UTF-8?q?feat(autopilot):=20add=20Staleness=20Sentin?= =?UTF-8?q?el=20=E2=80=94=20automated=20PR=20aging=20&=20escalation=20pipe?= =?UTF-8?q?line?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + autopilot/autopilot.py | 32 ++ autopilot/config.yaml | 23 ++ autopilot/staleness_engine.py | 516 ++++++++++++++++++++++++++++ tests/unit/test_staleness_engine.py | 487 ++++++++++++++++++++++++++ 5 files changed, 1059 insertions(+) create mode 100644 autopilot/staleness_engine.py create mode 100644 tests/unit/test_staleness_engine.py diff --git a/.gitignore b/.gitignore index 75e39d7..a2c1006 100644 --- a/.gitignore +++ b/.gitignore @@ -84,6 +84,7 @@ landing/tsconfig.tsbuildinfo .llm_costs.jsonl coverage.json coverage.xml +autopilot/staleness_state.json # MCP server (local tooling, not part of the pipeline) mcp_server/ diff --git a/autopilot/autopilot.py b/autopilot/autopilot.py index 3def661..e269375 100644 --- a/autopilot/autopilot.py +++ b/autopilot/autopilot.py @@ -291,6 +291,33 @@ def generate_summary(self, priorities: list[dict]) -> str: return "\n".join(summary) + def _run_staleness_check(self, staleness_cfg: dict) -> None: + """Invoke StalenessEngine as part of the daily run.""" + try: + from autopilot.staleness_engine import StalenessEngine + + state_file = staleness_cfg.get("state_file") + engine = StalenessEngine(config=staleness_cfg, state_file=state_file) + + # On very first run the state file won't exist — seed it silently. + if not engine.state_file.exists(): + print("[staleness] First run detected — seeding state (no comments posted).") + engine.init_state_from_prs(self.repo_data) + return + + actions = engine.process_prs(self.repo_data) + posted = sum(1 for a in actions if a.get("posted")) + dry_run_count = sum(1 for a in actions if a.get("dry_run")) + print( + f"[staleness] {len(actions)} escalation(s) processed " + f"(posted={posted}, dry_run={dry_run_count})" + ) + except Exception as exc: + print( + f"[staleness] WARNING: staleness check failed (non-fatal, daily run continues): {exc}", + file=sys.stderr, + ) + def run(self, output_file: str | None = None) -> str: """Main execution: fetch data, analyze, generate summary""" print("=" * 60) @@ -346,6 +373,11 @@ def run(self, output_file: str | None = None) -> str: f.write(summary) print(f"✅ Summary written to: {output_file_path}") + # Optional: run staleness sentinel if enabled in config + staleness_cfg = self.config.get("staleness", {}) + if staleness_cfg.get("enabled", False): + self._run_staleness_check(staleness_cfg) + runtime = time.time() - self.start_time print(f"\n✅ Complete in {runtime:.1f}s") diff --git a/autopilot/config.yaml b/autopilot/config.yaml index cd16a5c..3c884f0 100644 --- a/autopilot/config.yaml +++ b/autopilot/config.yaml @@ -71,3 +71,26 @@ api: rate_limit_buffer: 100 # Keep this many API calls in reserve timeout: 30 # seconds retry_attempts: 3 + +# Staleness Sentinel — PR Aging & Escalation +# Set ENABLE_LIVE_MODE=true in the environment to post real GitHub comments. +# Dry-run mode is the default: actions are logged but no comments are posted. +staleness: + enabled: true + + # Maximum number of Claude LLM calls per run for comment generation. + # Override with MAX_CLAUDE_CALLS_PER_RUN env var. + max_claude_calls_per_run: 20 + + # Tier thresholds (days open). Matches TIERS in staleness_engine.py. + # T0 (0-6d): fresh — no action + # T1 (7-29d): aging — informational comment + # T2 (30-89d): stale — escalation comment + # T3 (90+d): critical — urgent escalation comment + tiers: + T1_days: 7 + T2_days: 30 + T3_days: 90 + + # Path to the state file that persists per-PR tier assignments. + state_file: autopilot/staleness_state.json diff --git a/autopilot/staleness_engine.py b/autopilot/staleness_engine.py new file mode 100644 index 0000000..7e36bde --- /dev/null +++ b/autopilot/staleness_engine.py @@ -0,0 +1,516 @@ +#!/usr/bin/env python3 +""" +Staleness Sentinel — Automated PR Aging & Escalation Pipeline + +Monitors open PRs across configured repositories, assigns age-based tier +classifications, and posts escalation comments via GitHub when a PR moves +to a higher-severity tier. + +Safety defaults: +- Dry-run is ON by default. Set ``ENABLE_LIVE_MODE=true`` in the environment + to allow write operations (GitHub PR comments + state file writes). +- Claude call cap defaults to 20 per run. Override with the + ``MAX_CLAUDE_CALLS_PER_RUN`` env var or ``staleness.max_claude_calls_per_run`` + config key. + +Usage (standalone): + python -m autopilot.staleness_engine [--config config.yaml] [--init-only] +""" + +from __future__ import annotations + +import json +import logging +import os +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from github import GithubException + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Tier definitions +# Format: tier_key -> (min_days_inclusive, max_days_inclusive_or_None, label, emoji) +# --------------------------------------------------------------------------- +TIERS: dict[str, tuple[int, int | None, str, str]] = { + "T0": (0, 6, "fresh", "🟢"), + "T1": (7, 29, "aging", "🟡"), + "T2": (30, 89, "stale", "🟠"), + "T3": (90, None, "critical", "🔴"), +} + +_TIER_ORDER: list[str] = ["T0", "T1", "T2", "T3"] +_DEFAULT_STATE_FILE = Path(__file__).parent / "staleness_state.json" +_SCHEMA_VERSION = 1 + + +def classify_tier(days_open: int) -> str: + """Return the staleness tier key (T0–T3) for a PR open *days_open* days.""" + for tier_key in reversed(_TIER_ORDER): + min_days, _, _, _ = TIERS[tier_key] + if days_open >= min_days: + return tier_key + return "T0" + + +def _utcnow() -> str: + """Return current UTC time as ISO-8601 string.""" + return datetime.now(tz=timezone.utc).isoformat() + + +def _days_open(pr: Any) -> int: + """Return the number of days a PR has been open (timezone-naive arithmetic).""" + return (datetime.now() - pr.created_at.replace(tzinfo=None)).days + + +class StalenessEngine: + """ + Cross-repo PR staleness monitor with tiered escalation. + + Args: + config: Staleness-specific config dict (from ``staleness`` section + of ``autopilot/config.yaml``). + state_file: Path to the JSON file that persists per-PR tier state. + Defaults to ``autopilot/staleness_state.json``. + llm_config: Optional dict forwarded to ``LLMClient`` for Claude-powered + comment generation. When omitted, comments fall back to + static templates. + """ + + def __init__( + self, + config: dict[str, Any] | None = None, + state_file: str | Path | None = None, + llm_config: dict[str, Any] | None = None, + ) -> None: + self.config: dict[str, Any] = config or {} + self.state_file = Path(state_file) if state_file else _DEFAULT_STATE_FILE + self.llm_config = llm_config + + # Dry-run: write operations are suppressed unless ENABLE_LIVE_MODE=true + self.dry_run: bool = ( + os.getenv("ENABLE_LIVE_MODE", "").strip().lower() != "true" + ) + + # Claude call cap (env var takes precedence over config) + env_cap = os.getenv("MAX_CLAUDE_CALLS_PER_RUN", "").strip() + cfg_cap = self.config.get("max_claude_calls_per_run", 20) + self.max_claude_calls: int = int(env_cap) if env_cap.isdigit() else int(cfg_cap) + self._claude_calls_this_run: int = 0 + + self._state: dict[str, Any] = self._load_state() + + # ------------------------------------------------------------------ + # State persistence + # ------------------------------------------------------------------ + + def _load_state(self) -> dict[str, Any]: + """Load staleness state from disk; return empty state on miss or error.""" + if self.state_file.exists(): + try: + with open(self.state_file, encoding="utf-8") as f: + data = json.load(f) + if data.get("schema_version") == _SCHEMA_VERSION: + return data + logger.warning( + "staleness_state.json schema mismatch (expected v%d) — resetting.", + _SCHEMA_VERSION, + ) + except (json.JSONDecodeError, OSError) as exc: + logger.warning("Failed to load staleness state: %s", exc) + return {"schema_version": _SCHEMA_VERSION, "prs": {}} + + def _save_state(self) -> None: + """Write current state to disk. No-op in dry-run mode.""" + if self.dry_run: + logger.debug("[dry-run] _save_state: skipping state file write.") + return + self._state["saved_at"] = _utcnow() + try: + with open(self.state_file, "w", encoding="utf-8") as f: + json.dump(self._state, f, indent=2) + except OSError as exc: + logger.error("Failed to save staleness state: %s", exc) + + # ------------------------------------------------------------------ + # Public API + # ------------------------------------------------------------------ + + def get_state(self) -> dict[str, Any]: + """Return a snapshot of the current in-memory state.""" + import copy + + return copy.deepcopy(self._state) + + def classify_tier(self, days_open: int) -> str: + """Classify a PR by the number of days it has been open.""" + return classify_tier(days_open) + + def init_state_from_prs(self, repo_data: dict[str, Any]) -> None: + """ + Pre-populate state for all currently open PRs **without** posting + any escalation comments. + + Call this on first run to prevent comment-spam on PRs that have + already been open a long time before the sentinel was deployed. + + Args: + repo_data: The ``repo_data`` dict produced by + ``GitHubAutopilot.fetch_all_repos()``. + """ + now = _utcnow() + seeded = 0 + for repo_key, data in repo_data.items(): + for pr in data.get("pulls", []): + days_open = _days_open(pr) + tier = classify_tier(days_open) + pr_key = f"{repo_key}/{pr.number}" + if pr_key not in self._state["prs"]: + self._state["prs"][pr_key] = { + "tier": tier, + "days_open": days_open, + "pr_number": pr.number, + "pr_title": pr.title, + "pr_url": pr.html_url, + "repo": repo_key, + "first_seen": now, + "last_escalated_tier": tier, + "last_comment_at": None, + "comment_count": 0, + } + seeded += 1 + + self._save_state() + logger.info( + "[staleness] init_state_from_prs: seeded %d new PR entries (dry_run=%s).", + seeded, + self.dry_run, + ) + + def process_prs( + self, + repo_data: dict[str, Any], + ) -> list[dict[str, Any]]: + """ + Scan all open PRs, refresh tier assignments, and escalate when a PR + moves to a higher-severity tier. + + Args: + repo_data: The ``repo_data`` dict produced by + ``GitHubAutopilot.fetch_all_repos()``. + + Returns: + A list of action records — one per escalated PR. Each record + contains ``pr_key``, ``tier``, ``days_open``, ``comment_body``, + ``posted``, and ``dry_run`` fields. + """ + actions: list[dict[str, Any]] = [] + + for repo_key, data in repo_data.items(): + repo_obj = data.get("repo_obj") + for pr in data.get("pulls", []): + days_open = _days_open(pr) + current_tier = classify_tier(days_open) + pr_key = f"{repo_key}/{pr.number}" + + entry = self._state["prs"].get(pr_key) + if entry is None: + # First time we see this PR — seed without escalating + self._state["prs"][pr_key] = { + "tier": current_tier, + "days_open": days_open, + "pr_number": pr.number, + "pr_title": pr.title, + "pr_url": pr.html_url, + "repo": repo_key, + "first_seen": _utcnow(), + "last_escalated_tier": current_tier, + "last_comment_at": None, + "comment_count": 0, + } + continue + + # Refresh current values + entry["tier"] = current_tier + entry["days_open"] = days_open + + prev_tier = entry.get("last_escalated_tier", entry["tier"]) + if _TIER_ORDER.index(current_tier) > _TIER_ORDER.index(prev_tier): + action = self._escalate( + pr_key, pr, current_tier, days_open, repo_obj + ) + actions.append(action) + + self._save_state() + return actions + + # ------------------------------------------------------------------ + # Escalation helpers + # ------------------------------------------------------------------ + + def _escalate( + self, + pr_key: str, + pr: Any, + tier: str, + days_open: int, + repo_obj: Any, + ) -> dict[str, Any]: + """Build, optionally post, and return an action record for one PR.""" + tier_label = TIERS[tier][2] + tier_emoji = TIERS[tier][3] + comment_body = self._generate_comment(pr, tier, tier_label, tier_emoji, days_open) + + action: dict[str, Any] = { + "pr_key": pr_key, + "pr_number": pr.number, + "pr_title": pr.title, + "pr_url": pr.html_url, + "tier": tier, + "days_open": days_open, + "comment_body": comment_body, + "posted": False, + "dry_run": self.dry_run, + } + + if self.dry_run: + logger.info( + "[dry-run] Would escalate %s to %s (%s, %d days): %s", + pr_key, + tier, + tier_label, + days_open, + pr.title, + ) + else: + posted = self._post_comment(repo_obj, pr, comment_body) + action["posted"] = posted + if posted: + entry = self._state["prs"][pr_key] + entry["last_escalated_tier"] = tier + entry["last_comment_at"] = _utcnow() + entry["comment_count"] = entry.get("comment_count", 0) + 1 + + return action + + def _generate_comment( + self, + pr: Any, + tier: str, + tier_label: str, + tier_emoji: str, + days_open: int, + ) -> str: + """ + Return a comment body string. + + Tries Claude first (if ``llm_config`` is set and the per-run call cap + has not been exhausted); falls back to a static template. + """ + if self.llm_config and self._claude_calls_this_run < self.max_claude_calls: + claude_text = self._generate_claude_comment( + pr, tier, tier_label, tier_emoji, days_open + ) + if claude_text: + return claude_text + + return self._static_comment(pr, tier, tier_label, tier_emoji, days_open) + + def _generate_claude_comment( + self, + pr: Any, + tier: str, + tier_label: str, + tier_emoji: str, + days_open: int, + ) -> str | None: + """Call Claude to generate a contextual escalation comment.""" + import asyncio + + try: + from core.llm_provider import LLMClient + except ImportError: + logger.warning( + "core.llm_provider not available — using static template." + ) + return None + + if self._claude_calls_this_run >= self.max_claude_calls: + logger.info( + "[staleness] Claude call cap reached (%d/%d) — using static template.", + self._claude_calls_this_run, + self.max_claude_calls, + ) + return None + + system = ( + "You are a helpful engineering assistant writing concise GitHub PR " + "escalation comments. Be professional, constructive, and brief " + "(≤120 words). Do not include greetings." + ) + prompt = ( + f"Write a PR escalation comment for PR #{pr.number}: '{pr.title}'. " + f"It has been open for {days_open} days and reached staleness tier " + f"{tier} ({tier_label} {tier_emoji}). " + "Remind the team it needs attention and suggest a concrete next step." + ) + + try: + client = LLMClient(self.llm_config) + result = asyncio.run( + client.generate(prompt, system_prompt=system, max_tokens=200) + ) + self._claude_calls_this_run += 1 + logger.info( + "[staleness] Claude call %d/%d completed. Session cost: %s", + self._claude_calls_this_run, + self.max_claude_calls, + client.get_session_cost(), + ) + return result.get("content", "").strip() or None + except Exception as exc: + logger.warning("Claude comment generation failed: %s", exc) + return None + + @staticmethod + def _static_comment( + pr: Any, + tier: str, + tier_label: str, + tier_emoji: str, + days_open: int, + ) -> str: + """Return a static escalation comment template for the given tier.""" + templates: dict[str, str] = { + "T1": ( + "👋 This PR has been open for **{days} days** and is now " + "marked **aging** {emoji}. Please review at your earliest " + "convenience." + ), + "T2": ( + "⚠️ This PR has been open for **{days} days** and is now " + "marked **stale** {emoji}. It needs attention — please review, " + "update, or close this PR to keep the backlog healthy." + ), + "T3": ( + "🚨 This PR has been open for **{days} days** and is now " + "**critically stale** {emoji}. Immediate action is required — " + "please merge, close, or leave a status update." + ), + } + template = templates.get( + tier, + "This PR has reached staleness tier {tier} ({label} {emoji}) " + "after {days} days.", + ) + return template.format( + days=days_open, + emoji=tier_emoji, + tier=tier, + label=tier_label, + ) + + def _post_comment(self, repo_obj: Any, pr: Any, comment: str) -> bool: + """Post *comment* to the GitHub PR issue thread.""" + if repo_obj is None: + logger.warning( + "[staleness] repo_obj is None for PR #%d — skipping comment.", + pr.number, + ) + return False + try: + pull = repo_obj.get_pull(pr.number) + pull.create_issue_comment(comment) + logger.info( + "[staleness] Posted escalation comment on %s PR #%d.", + pr.html_url, + pr.number, + ) + return True + except GithubException as exc: + logger.error( + "[staleness] Failed to post comment on PR #%d: %s", + pr.number, + exc, + ) + return False + + +def main() -> None: + """CLI entry point for standalone staleness checks.""" + import argparse + import sys + + from dotenv import load_dotenv + + load_dotenv() + + parser = argparse.ArgumentParser( + description="Staleness Sentinel — PR Aging & Escalation" + ) + parser.add_argument( + "--config", + default="config.yaml", + help="Autopilot config file (default: config.yaml)", + ) + parser.add_argument( + "--state-file", + default=str(_DEFAULT_STATE_FILE), + help="Path to staleness_state.json", + ) + parser.add_argument( + "--init-only", + action="store_true", + help=( + "Seed state file from current open PRs without posting any comments. " + "Run this once before going live to avoid comment-spam on old PRs." + ), + ) + args = parser.parse_args() + + from autopilot.autopilot import GitHubAutopilot + + try: + autopilot = GitHubAutopilot(config_path=args.config) + autopilot.fetch_all_repos() + + staleness_cfg = autopilot.config.get("staleness", {}) + engine = StalenessEngine( + config=staleness_cfg, + state_file=args.state_file, + ) + + if args.init_only: + engine.init_state_from_prs(autopilot.repo_data) + state = engine.get_state() + print( + f"✅ State seeded: {len(state['prs'])} PRs recorded " + f"(dry_run={engine.dry_run}, no comments posted)" + ) + return + + actions = engine.process_prs(autopilot.repo_data) + posted = sum(1 for a in actions if a.get("posted")) + dry_run_logged = sum(1 for a in actions if a.get("dry_run")) + print( + f"✅ Staleness check complete: {len(actions)} escalation(s) " + f"(posted={posted}, dry_run_logged={dry_run_logged})" + ) + + for action in actions: + tier_label = TIERS[action["tier"]][2] + status = "DRY-RUN" if action["dry_run"] else ("✓" if action["posted"] else "✗") + print( + f" [{status}] {action['pr_key']} → {action['tier']} " + f"({tier_label}, {action['days_open']}d): {action['pr_title']}" + ) + + except Exception as exc: + print(f"❌ Error: {exc}", file=sys.stderr) + sys.exit(1) + + +if __name__ == "__main__": + main() diff --git a/tests/unit/test_staleness_engine.py b/tests/unit/test_staleness_engine.py new file mode 100644 index 0000000..0036e52 --- /dev/null +++ b/tests/unit/test_staleness_engine.py @@ -0,0 +1,487 @@ +"""Unit tests for autopilot/staleness_engine.py""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +def _make_pr(number: int = 1, days_old: int = 5, title: str = "Test PR") -> MagicMock: + pr = MagicMock() + pr.number = number + pr.title = title + pr.html_url = f"https://github.com/acme/repo/pull/{number}" + pr.created_at = MagicMock() + pr.created_at.replace.return_value = datetime.now() - timedelta(days=days_old) + return pr + + +def _make_repo_data(pulls: list | None = None) -> dict: + return { + "acme/repo": { + "pulls": pulls or [], + "issues": [], + "commits": [], + "repo_obj": MagicMock(), + "priority": "critical", + } + } + + +def _make_engine(tmp_path: Path, env: dict | None = None, config: dict | None = None): + from autopilot.staleness_engine import StalenessEngine + + state_file = tmp_path / "staleness_state.json" + env = env or {} + with patch.dict("os.environ", env, clear=False): + return StalenessEngine( + config=config or {}, + state_file=state_file, + ) + + +# --------------------------------------------------------------------------- +# classify_tier (module-level function) +# --------------------------------------------------------------------------- + +class TestClassifyTier: + def test_zero_days_is_t0(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(0) == "T0" + + def test_six_days_is_t0(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(6) == "T0" + + def test_seven_days_is_t1(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(7) == "T1" + + def test_twenty_nine_days_is_t1(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(29) == "T1" + + def test_thirty_days_is_t2(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(30) == "T2" + + def test_eighty_nine_days_is_t2(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(89) == "T2" + + def test_ninety_days_is_t3(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(90) == "T3" + + def test_one_hundred_days_is_t3(self): + from autopilot.staleness_engine import classify_tier + assert classify_tier(108) == "T3" + + +# --------------------------------------------------------------------------- +# StalenessEngine.__init__ / dry-run flag +# --------------------------------------------------------------------------- + +class TestDryRunFlag: + def test_dry_run_true_by_default(self, tmp_path): + engine = _make_engine(tmp_path, env={}) + assert engine.dry_run is True + + def test_dry_run_false_when_enable_live_mode_true(self, tmp_path): + engine = _make_engine(tmp_path, env={"ENABLE_LIVE_MODE": "true"}) + assert engine.dry_run is False + + def test_dry_run_still_true_for_other_values(self, tmp_path): + for val in ("1", "yes", ""): + engine = _make_engine(tmp_path, env={"ENABLE_LIVE_MODE": val}) + assert engine.dry_run is True, f"Expected dry_run=True for ENABLE_LIVE_MODE={val!r}" + + def test_dry_run_false_for_uppercase_true(self, tmp_path): + engine = _make_engine(tmp_path, env={"ENABLE_LIVE_MODE": "TRUE"}) + assert engine.dry_run is False + + +# --------------------------------------------------------------------------- +# Claude call cap +# --------------------------------------------------------------------------- + +class TestClaudeCallCap: + def test_default_cap_is_20(self, tmp_path): + engine = _make_engine(tmp_path) + assert engine.max_claude_calls == 20 + + def test_env_var_overrides_cap(self, tmp_path): + engine = _make_engine(tmp_path, env={"MAX_CLAUDE_CALLS_PER_RUN": "5"}) + assert engine.max_claude_calls == 5 + + def test_config_key_sets_cap(self, tmp_path): + engine = _make_engine(tmp_path, config={"max_claude_calls_per_run": 10}) + assert engine.max_claude_calls == 10 + + def test_env_var_takes_precedence_over_config(self, tmp_path): + engine = _make_engine( + tmp_path, + env={"MAX_CLAUDE_CALLS_PER_RUN": "3"}, + config={"max_claude_calls_per_run": 15}, + ) + assert engine.max_claude_calls == 3 + + +# --------------------------------------------------------------------------- +# State persistence +# --------------------------------------------------------------------------- + +class TestStatePersistence: + def test_empty_state_when_no_file(self, tmp_path): + engine = _make_engine(tmp_path) + state = engine.get_state() + assert state["schema_version"] == 1 + assert state["prs"] == {} + + def test_loads_existing_state_file(self, tmp_path): + state_file = tmp_path / "staleness_state.json" + state_file.write_text( + json.dumps({"schema_version": 1, "prs": {"acme/repo/42": {"tier": "T2"}}}) + ) + from autopilot.staleness_engine import StalenessEngine + + engine = StalenessEngine(state_file=state_file) + assert "acme/repo/42" in engine.get_state()["prs"] + + def test_resets_state_on_schema_mismatch(self, tmp_path): + state_file = tmp_path / "staleness_state.json" + state_file.write_text(json.dumps({"schema_version": 99, "prs": {"x": {}}})) + from autopilot.staleness_engine import StalenessEngine + + engine = StalenessEngine(state_file=state_file) + assert engine.get_state()["prs"] == {} + + def test_save_state_noop_in_dry_run(self, tmp_path): + engine = _make_engine(tmp_path, env={}) + assert engine.dry_run is True + engine._save_state() + assert not engine.state_file.exists() + + def test_save_state_writes_file_in_live_mode(self, tmp_path): + engine = _make_engine(tmp_path, env={"ENABLE_LIVE_MODE": "true"}) + engine._save_state() + assert engine.state_file.exists() + data = json.loads(engine.state_file.read_text()) + assert data["schema_version"] == 1 + + +# --------------------------------------------------------------------------- +# init_state_from_prs +# --------------------------------------------------------------------------- + +class TestInitStateFromPrs: + def test_seeds_prs_without_posting_comments(self, tmp_path): + engine = _make_engine(tmp_path) + pr = _make_pr(number=10, days_old=100) + repo_data = _make_repo_data(pulls=[pr]) + + engine.init_state_from_prs(repo_data) + + state = engine.get_state() + assert "acme/repo/10" in state["prs"] + entry = state["prs"]["acme/repo/10"] + assert entry["tier"] == "T3" + assert entry["last_comment_at"] is None + assert entry["comment_count"] == 0 + + def test_does_not_overwrite_existing_entry(self, tmp_path): + state_file = tmp_path / "staleness_state.json" + state_file.write_text( + json.dumps({ + "schema_version": 1, + "prs": { + "acme/repo/10": { + "tier": "T2", + "last_escalated_tier": "T2", + "comment_count": 1, + } + }, + }) + ) + from autopilot.staleness_engine import StalenessEngine + + engine = StalenessEngine(state_file=state_file) + pr = _make_pr(number=10, days_old=100) + engine.init_state_from_prs({"acme/repo": {"pulls": [pr]}}) + + entry = engine.get_state()["prs"]["acme/repo/10"] + # Original entry preserved — not overwritten + assert entry["tier"] == "T2" + assert entry["comment_count"] == 1 + + def test_state_file_not_written_in_dry_run(self, tmp_path): + engine = _make_engine(tmp_path, env={}) + pr = _make_pr(number=5, days_old=40) + engine.init_state_from_prs(_make_repo_data(pulls=[pr])) + assert not engine.state_file.exists() + + +# --------------------------------------------------------------------------- +# process_prs +# --------------------------------------------------------------------------- + +class TestProcessPrs: + def test_no_actions_when_no_prs(self, tmp_path): + engine = _make_engine(tmp_path) + actions = engine.process_prs(_make_repo_data(pulls=[])) + assert actions == [] + + def test_new_pr_seeded_without_escalation(self, tmp_path): + engine = _make_engine(tmp_path) + pr = _make_pr(number=1, days_old=50) + actions = engine.process_prs(_make_repo_data(pulls=[pr])) + # Brand-new PR — no escalation on first observation + assert actions == [] + assert "acme/repo/1" in engine.get_state()["prs"] + + def test_tier_increase_triggers_escalation(self, tmp_path): + state_file = tmp_path / "staleness_state.json" + state_file.write_text( + json.dumps({ + "schema_version": 1, + "prs": { + "acme/repo/7": { + "tier": "T1", + "days_open": 20, + "last_escalated_tier": "T1", + "last_comment_at": None, + "comment_count": 0, + } + }, + }) + ) + from autopilot.staleness_engine import StalenessEngine + + engine = StalenessEngine(state_file=state_file) + # PR is now 35 days old → T2 + pr = _make_pr(number=7, days_old=35) + actions = engine.process_prs(_make_repo_data(pulls=[pr])) + assert len(actions) == 1 + assert actions[0]["tier"] == "T2" + assert actions[0]["dry_run"] is True + assert actions[0]["posted"] is False + + def test_no_escalation_when_tier_unchanged(self, tmp_path): + state_file = tmp_path / "staleness_state.json" + state_file.write_text( + json.dumps({ + "schema_version": 1, + "prs": { + "acme/repo/3": { + "tier": "T2", + "days_open": 31, + "last_escalated_tier": "T2", + "last_comment_at": None, + "comment_count": 0, + } + }, + }) + ) + from autopilot.staleness_engine import StalenessEngine + + engine = StalenessEngine(state_file=state_file) + pr = _make_pr(number=3, days_old=40) # Still T2 + actions = engine.process_prs(_make_repo_data(pulls=[pr])) + assert actions == [] + + def test_no_escalation_when_tier_decreases(self, tmp_path): + """Tier should never decrease for open PRs, but guard against it.""" + state_file = tmp_path / "staleness_state.json" + state_file.write_text( + json.dumps({ + "schema_version": 1, + "prs": { + "acme/repo/9": { + "tier": "T3", + "days_open": 100, + "last_escalated_tier": "T3", + "last_comment_at": None, + "comment_count": 0, + } + }, + }) + ) + from autopilot.staleness_engine import StalenessEngine + + engine = StalenessEngine(state_file=state_file) + pr = _make_pr(number=9, days_old=10) # Would be T1, but last_escalated was T3 + actions = engine.process_prs(_make_repo_data(pulls=[pr])) + assert actions == [] + + +# --------------------------------------------------------------------------- +# Static comment templates +# --------------------------------------------------------------------------- + +class TestStaticComment: + def _make_pr_obj(self, number=1): + pr = MagicMock() + pr.number = number + pr.title = "Add feature X" + pr.html_url = f"https://github.com/acme/repo/pull/{number}" + return pr + + def test_t1_comment_mentions_aging(self, tmp_path): + from autopilot.staleness_engine import StalenessEngine + + engine = _make_engine(tmp_path) + pr = self._make_pr_obj() + comment = engine._static_comment(pr, "T1", "aging", "🟡", 10) + assert "aging" in comment.lower() or "🟡" in comment + assert "10" in comment + + def test_t2_comment_mentions_stale(self, tmp_path): + from autopilot.staleness_engine import StalenessEngine + + engine = _make_engine(tmp_path) + pr = self._make_pr_obj() + comment = engine._static_comment(pr, "T2", "stale", "🟠", 45) + assert "stale" in comment.lower() or "⚠" in comment + assert "45" in comment + + def test_t3_comment_mentions_critical(self, tmp_path): + from autopilot.staleness_engine import StalenessEngine + + engine = _make_engine(tmp_path) + pr = self._make_pr_obj() + comment = engine._static_comment(pr, "T3", "critical", "🔴", 100) + assert "critical" in comment.lower() or "🚨" in comment + assert "100" in comment + + +# --------------------------------------------------------------------------- +# _post_comment — dry-run safety +# --------------------------------------------------------------------------- + +class TestPostComment: + def test_returns_false_when_repo_obj_none(self, tmp_path): + engine = _make_engine(tmp_path, env={"ENABLE_LIVE_MODE": "true"}) + pr = _make_pr() + result = engine._post_comment(None, pr, "hello") + assert result is False + + def test_posts_comment_in_live_mode(self, tmp_path): + engine = _make_engine(tmp_path, env={"ENABLE_LIVE_MODE": "true"}) + mock_repo = MagicMock() + mock_pull = MagicMock() + mock_repo.get_pull.return_value = mock_pull + pr = _make_pr(number=42) + + result = engine._post_comment(mock_repo, pr, "Escalation notice") + + assert result is True + mock_repo.get_pull.assert_called_once_with(42) + mock_pull.create_issue_comment.assert_called_once_with("Escalation notice") + + def test_returns_false_on_github_exception(self, tmp_path): + from github import GithubException + + engine = _make_engine(tmp_path, env={"ENABLE_LIVE_MODE": "true"}) + mock_repo = MagicMock() + mock_repo.get_pull.side_effect = GithubException(403, {"message": "Forbidden"}) + pr = _make_pr(number=5) + + result = engine._post_comment(mock_repo, pr, "notice") + assert result is False + + +# --------------------------------------------------------------------------- +# _generate_comment — falls back to static when llm_config absent +# --------------------------------------------------------------------------- + +class TestGenerateComment: + def test_uses_static_when_no_llm_config(self, tmp_path): + engine = _make_engine(tmp_path) + assert engine.llm_config is None + pr = _make_pr(number=1, days_old=95) + comment = engine._generate_comment(pr, "T3", "critical", "🔴", 95) + assert "95" in comment + + def test_uses_static_when_cap_reached(self, tmp_path): + engine = _make_engine(tmp_path, config={"max_claude_calls_per_run": 0}) + engine.llm_config = {"llm_provider": "anthropic"} # set but cap=0 + pr = _make_pr(number=2, days_old=40) + comment = engine._generate_comment(pr, "T2", "stale", "🟠", 40) + assert "40" in comment + + +# --------------------------------------------------------------------------- +# Escalate — dry_run suppresses posting +# --------------------------------------------------------------------------- + +class TestEscalate: + def _setup_engine_with_entry(self, tmp_path, pr_key: str, last_tier: str): + state_file = tmp_path / "staleness_state.json" + state_file.write_text( + json.dumps({ + "schema_version": 1, + "prs": { + pr_key: { + "tier": last_tier, + "days_open": 20, + "last_escalated_tier": last_tier, + "last_comment_at": None, + "comment_count": 0, + } + }, + }) + ) + from autopilot.staleness_engine import StalenessEngine + + return StalenessEngine(state_file=state_file) + + def test_dry_run_does_not_post_comment(self, tmp_path): + engine = self._setup_engine_with_entry(tmp_path, "acme/repo/1", "T1") + assert engine.dry_run is True + pr = _make_pr(number=1, days_old=35) + mock_repo = MagicMock() + + action = engine._escalate("acme/repo/1", pr, "T2", 35, mock_repo) + + assert action["posted"] is False + assert action["dry_run"] is True + mock_repo.get_pull.assert_not_called() + + def test_live_mode_posts_and_updates_entry(self, tmp_path): + engine = self._setup_engine_with_entry(tmp_path, "acme/repo/2", "T1") + engine.dry_run = False # Force live mode directly + pr = _make_pr(number=2, days_old=35) + mock_repo = MagicMock() + mock_pull = MagicMock() + mock_repo.get_pull.return_value = mock_pull + + action = engine._escalate("acme/repo/2", pr, "T2", 35, mock_repo) + + assert action["posted"] is True + mock_pull.create_issue_comment.assert_called_once() + + entry = engine.get_state()["prs"]["acme/repo/2"] + assert entry["last_escalated_tier"] == "T2" + assert entry["comment_count"] == 1 + assert entry["last_comment_at"] is not None + + +# --------------------------------------------------------------------------- +# StalenessEngine.classify_tier method delegates to module function +# --------------------------------------------------------------------------- + +class TestClassifyTierMethod: + def test_method_matches_module_function(self, tmp_path): + from autopilot.staleness_engine import classify_tier + + engine = _make_engine(tmp_path) + for days in [0, 6, 7, 29, 30, 89, 90, 120]: + assert engine.classify_tier(days) == classify_tier(days)