diff --git a/docs/flows.md b/docs/flows.md index 88ece4e36..483d5d455 100644 --- a/docs/flows.md +++ b/docs/flows.md @@ -113,3 +113,10 @@ cao schedule run daily-standup # Remove a flow cao schedule remove daily-standup ``` + +## Workflow scheduling example + +[`examples/workflows/pr-health/`](../examples/workflows/pr-health/) combines a +parameterized Python workflow with a scheduled flow. Its weekly cron entry uses +a deterministic guard to produce an exact 14-day cadence across month +boundaries. Scheduled runs default to non-mutating dry-run mode. diff --git a/docs/workflows.md b/docs/workflows.md index 80050db28..63a86c3f5 100644 --- a/docs/workflows.md +++ b/docs/workflows.md @@ -244,4 +244,7 @@ All eight verbs live under `cao workflow`. - [`conditional_example.py`](examples/conditional_example.py) — branching, explicit `step_id` per branch. - [`fanout_example.py`](examples/fanout_example.py) — concurrent fan-out via `ThreadPoolExecutor`. - [`loop_raw_http_example.py`](examples/loop_raw_http_example.py) — the same loop with no shim, raw `urllib` against the identity env vars. +- [`examples/workflows/pr-health/`](../examples/workflows/pr-health/) — a + deterministic open-PR health scorer with guarded enforcement and a biweekly + scheduled-flow example. - [`skills/cao-workflow/SKILL.md`](../skills/cao-workflow/SKILL.md) — the agent-facing skill that teaches this lifecycle. diff --git a/examples/workflows/pr-health/README.md b/examples/workflows/pr-health/README.md new file mode 100644 index 000000000..9608d2146 --- /dev/null +++ b/examples/workflows/pr-health/README.md @@ -0,0 +1,194 @@ +# Pull Request Health Workflow + +This example evaluates every open GitHub pull request with deterministic rules, +tracks degradation across runs, and produces an auditable Markdown and JSON +report. It also includes a CAO scheduled flow with an exact 14-day cadence. + +The score is rule-based. The optional reviewer agent may summarize importance, +but it cannot change scores, categories, next-actor attribution, or actions. Its +prompt reads untrusted PR text (titles, bodies, comments), so treat its output as +advisory prose only — nothing it says can alter a score or trigger an action. + +## Safety model + +The workflow defaults to `dry_run` and makes no GitHub changes. Apply mode has +additional safeguards: + +- A PR must score below 60 on two separate dated evaluations before an owner + warning is eligible. +- A score at or below 50 can move a PR to draft only after a warning has been + unanswered for at least seven days. +- A PR can become a closure candidate only after another 14 days without owner + activity and a final score below 30. +- P0/P1 and approved PRs are escalated instead of drafted or closed. +- Closure requires the PR number in the explicit `close_allowlist`. +- Every candidate is fetched and scored again immediately before mutation. +- Only hidden markers in comments authored by the authenticated workflow + identity can advance lifecycle stages or make actions idempotent. +- Each lifecycle stage notifies the owner at most once, ever. Idempotency keys + on the marker's stage, not its text, so a repeat run cannot re-post a + notification even though the marker embeds that run's score and date. +- Lifecycle progression does not depend on the run cadence. The + furthest-advanced marker owns the PR's grace period, so a weekly, biweekly, or + ad-hoc run reaches the same stage after the same elapsed time. +- A per-PR state problem (backfill, clock skew, a reopened PR carrying old + state) restarts that PR's observation streak; it never aborts the run for the + other PRs. +- Runs for the same repository are serialized with a local file lock. + +The dry-run and apply schedules are separate templates. Registering the apply +template is a deliberate standing authorization for future comments and draft +changes. Both templates keep `close_allowlist` empty, so unattended runs cannot +close PRs. Both may be registered together: the guard emits mode-qualified run +and snapshot identifiers so they never collide on a shared due date. + +> **Apply mode is a standing unattended write-grant.** Once registered, the +> apply schedule comments on and drafts other contributors' pull requests under +> the operator's `gh` identity, with no per-run review. Drafting someone's PR +> has real social impact. Trial the apply template against a fork you own before +> registering it against a shared repository, and read the dry-run report for at +> least one full cycle first. + +## Scoring + +| Dimension | Maximum | Deterministic signals | +| --- | ---: | --- | +| CI | 20 | passing, pending, missing, or failing checks | +| Mergeability | 15 | clean, blocked/behind, unknown, or conflicting | +| Review | 15 | approved, review required, draft, or changes requested | +| Engagement | 40 | days since the latest commit | +| Completeness | 10 | description, rationale, tests, and focused scope | + +Health bands are `healthy` (85-100), `active` (70-84), `watch` (60-69), +`at_risk` (51-59), `stalled` (30-50), and `abandoned` (0-29). +Priority is calculated separately so an unhealthy but important PR is escalated +rather than discarded. + +## Prerequisites + +- `cao-server` running +- `gh` installed and authenticated for the target repository +- CAO `developer` and `reviewer` profiles available +- A headless provider for the optional importance analysis + +## Install the workflow + +```bash +mkdir -p ~/.aws/cli-agent-orchestrator/workflows +install -m 0644 \ + examples/workflows/pr-health/pr_health.py \ + ~/.aws/cli-agent-orchestrator/workflows/pr_health.py + +cao workflow validate \ + ~/.aws/cli-agent-orchestrator/workflows/pr_health.py +``` + +## Run a dry evaluation + +Supply the date explicitly. The same inputs always select the same snapshot and +artifact directory, which keeps resume behavior deterministic. + +```bash +cao workflow run pr_health \ + --run-id pr-health-dry-2026-08-03 \ + --input repo=awslabs/cli-agent-orchestrator \ + --input as_of=2026-08-03 \ + --input snapshot_id=dry-2026-08-03 \ + --input importance_analysis=true \ + --input importance_provider=claude_code \ + --input importance_agent=reviewer \ + --input mode=dry_run \ + --json +``` + +Artifacts are written under: + +```text +~/.local/state/cao/pr-health//runs// +``` + +## Apply eligible actions + +Review the dry-run report first, then use a new run and snapshot ID: + +```bash +cao workflow run pr_health \ + --run-id pr-health-apply-2026-08-03 \ + --input repo=awslabs/cli-agent-orchestrator \ + --input as_of=2026-08-03 \ + --input snapshot_id=apply-2026-08-03 \ + --input importance_analysis=true \ + --input importance_provider=claude_code \ + --input importance_agent=reviewer \ + --input mode=apply \ + --input close_allowlist= \ + --json +``` + +An empty `close_allowlist` permits eligible comments and draft transitions but +prevents closure. To approve specific closures, pass a comma-separated list such +as `--input close_allowlist=123,456`. + +`importance_provider` and `importance_agent` select the headless CAO reviewer +step used for advisory importance synthesis. They do not affect deterministic +scores or actions. + +## Schedule every two weeks + +Traditional cron expressions cannot represent a continuous 14-day interval +across month boundaries. The included flow runs every Monday and uses +[`pr_health_biweekly_guard.py`](pr_health_biweekly_guard.py) to execute only on +dates exactly divisible by 14 from `ANCHOR`. + +1. Edit `REPOSITORY` and `ANCHOR` in the guard. +2. Choose either the dry-run template or the explicitly authorized apply + template. +3. Copy the chosen flow and guard to a durable local directory. +4. Register the chosen flow. + +```bash +mkdir -p ~/.cao/flows +install -m 0755 \ + examples/workflows/pr-health/pr_health_biweekly_guard.py \ + ~/.cao/flows/pr_health_biweekly_guard.py +install -m 0644 \ + examples/workflows/pr-health/pr-health-biweekly.md \ + ~/.cao/flows/pr-health-biweekly.md + +cao schedule add ~/.cao/flows/pr-health-biweekly.md +cao schedule list +``` + +For an apply schedule, install and register +`pr-health-biweekly-apply.md` instead — or in addition, since the guard +differentiates each mode's `run_id` and `snapshot_id`. Its empty closure +allowlist must remain empty for unattended operation. + +CAO uses APScheduler weekday numbering, where `0` is Monday. The flow therefore +uses `0 9 * * 0` for Monday at 09:00 in the server's local timezone. The +`cao-server` process must remain running for scheduled flows to execute. + +The guard derives `as_of` from the **UTC** date, not the server's local date, +because the scoring rules compare it against GitHub's UTC timestamps. A +local-date `as_of` would shift the 7/14/21-day threshold crossings by a day for +runs scheduled near midnight. + +Manage the schedule with: + +```bash +cao schedule disable pr-health-biweekly +cao schedule enable pr-health-biweekly +cao schedule remove pr-health-biweekly +``` + +## Files + +- [`pr_health.py`](pr_health.py): deterministic workflow and guarded enforcement +- [`pr-health-biweekly.md`](pr-health-biweekly.md): non-mutating scheduled flow +- [`pr-health-biweekly-apply.md`](pr-health-biweekly-apply.md): explicitly + authorized comment/draft scheduled flow +- [`pr_health_biweekly_guard.py`](pr_health_biweekly_guard.py): exact 14-day gate + +This example is a reference policy. Adjust thresholds and priority labels to +match the repository's contribution and maintainer policies before enabling +apply mode. diff --git a/examples/workflows/pr-health/pr-health-biweekly-apply.md b/examples/workflows/pr-health/pr-health-biweekly-apply.md new file mode 100644 index 000000000..f22f9acda --- /dev/null +++ b/examples/workflows/pr-health/pr-health-biweekly-apply.md @@ -0,0 +1,37 @@ +--- +name: pr-health-biweekly-apply +schedule: "0 9 * * 0" +agent_profile: developer +provider: codex +script: ./pr_health_biweekly_guard.py +--- + +> **STANDING UNATTENDED WRITE GRANT.** Registering this flow authorizes +> recurring, unattended writes to other people's pull requests — comments and +> draft transitions — under the operator's `gh` identity. Drafting a +> contributor's PR has real social impact. Trial this against a fork you own +> before registering it against a shared repository. + +This flow is an explicit standing authorization to apply eligible PR-health +comments and draft transitions on its scheduled runs. + +Run exactly this command and do not change, omit, or add arguments: + +```bash +cao workflow run pr_health --run-id [[run_id_apply]] \ + --input repo=[[repo]] \ + --input as_of=[[as_of]] \ + --input snapshot_id=[[snapshot_id_apply]] \ + --input importance_analysis=true \ + --input importance_provider=claude_code \ + --input importance_agent=reviewer \ + --input mode=apply \ + --input close_allowlist= \ + --json +``` + +Wait for the command to finish and report its structured result. Comments and +draft transitions are authorized only when the workflow's live revalidation +permits them. Closure is not authorized: the empty closure allowlist is +intentional and must remain empty. If the run ID already exists, inspect its +status instead of creating a duplicate run. diff --git a/examples/workflows/pr-health/pr-health-biweekly.md b/examples/workflows/pr-health/pr-health-biweekly.md new file mode 100644 index 000000000..6a2b13c2c --- /dev/null +++ b/examples/workflows/pr-health/pr-health-biweekly.md @@ -0,0 +1,29 @@ +--- +name: pr-health-biweekly +schedule: "0 9 * * 0" +agent_profile: developer +provider: codex +script: ./pr_health_biweekly_guard.py +--- + +This is a pre-authorized scheduled PR-health dry run. + +Run exactly this command and do not change, omit, or add arguments: + +```bash +cao workflow run pr_health --run-id [[run_id_dry_run]] \ + --input repo=[[repo]] \ + --input as_of=[[as_of]] \ + --input snapshot_id=[[snapshot_id_dry_run]] \ + --input importance_analysis=true \ + --input importance_provider=claude_code \ + --input importance_agent=reviewer \ + --input mode=dry_run \ + --input close_allowlist= \ + --json +``` + +Wait for the command to finish and report its structured result. Do not perform +any GitHub mutation. The empty closure allowlist is intentional and must remain +empty. If the run ID already exists, inspect its status instead of creating a +duplicate run. diff --git a/examples/workflows/pr-health/pr_health.py b/examples/workflows/pr-health/pr_health.py new file mode 100644 index 000000000..f33fa847a --- /dev/null +++ b/examples/workflows/pr-health/pr_health.py @@ -0,0 +1,1207 @@ +"""Deterministic health analysis and guarded enforcement for open GitHub PRs. + +Dry-run mode snapshots GitHub data, calculates scores with fixed rules, persists +the two-observation warning state, and asks a reviewer for an advisory importance +synthesis. Apply mode revalidates every candidate against live data before making +an idempotent comment or state change. Closure also requires an explicit allowlist. + +Two invariants keep enforcement from spamming or stalling, and both are covered +by ``test/examples/test_pr_health_workflow_example.py``: + +- Each lifecycle stage notifies the owner at most once. Idempotency keys on a + marker's *stage*, never on its full text, because the marker also embeds the + emitting run's score and ``as_of``. +- Lifecycle progression is independent of the run cadence. The + furthest-advanced marker owns the PR's grace period, so a later earlier-stage + comment cannot restart the clock or shadow the closure branch. + +Example (authoring does not authorize this run): + cao workflow run pr_health --run-id pr-health-2026-07-31 \ + --input as_of=2026-07-31 \ + --input snapshot_id=2026-07-31 +""" + +from __future__ import annotations + +import fcntl +import json +import re +import subprocess +from pathlib import Path +from typing import Any, TextIO +from urllib.parse import quote + +from cao_workflow import ShimError, emit_output, get_inputs, run_step + +INPUTS = { + "repo": { + "type": "string", + "required": False, + "default": "awslabs/cli-agent-orchestrator", + }, + "as_of": { + "type": "string", + "required": True, + }, + "snapshot_id": { + "type": "string", + "required": True, + }, + "max_prs": { + "type": "int", + "required": False, + "default": 500, + }, + "importance_analysis": { + "type": "bool", + "required": False, + "default": True, + }, + "importance_provider": { + "type": "string", + "required": False, + "default": "claude_code", + }, + "importance_agent": { + "type": "string", + "required": False, + "default": "reviewer", + }, + "mode": { + "type": "string", + "required": False, + "default": "dry_run", + }, + "close_allowlist": { + "type": "string", + "required": False, + "default": "", + }, +} + +SCHEMA_VERSION = 1 +MARKER_RE = re.compile( + r"" +) +ESCALATION_MARKER_RE = re.compile( + r"" +) +# Lifecycle stages never regress: the furthest-advanced stage owns the PR's +# grace period, so a later-posted earlier-stage marker cannot shadow it. +STAGE_RANK = {"warning": 0, "draft": 1, "closed": 2} +SAFE_ID_RE = re.compile(r"^[A-Za-z0-9._-]+$") +RESERVED_IDS = {".", ".."} +ISSUE_REF_RE = re.compile(r"(?i)(?:close[sd]?|fix(?:e[sd])?|resolve[sd]?)?\s*#\d+") +RATIONALE_RE = re.compile(r"(?i)\b(?:motivation|rationale|problem|why|because)\b") +TEST_RE = re.compile(r"(?i)\b(?:test|tests|tested|testing|verification)\b") + +FAILURE_CONCLUSIONS = { + "ACTION_REQUIRED", + "CANCELLED", + "ERROR", + "FAILURE", + "STALE", + "TIMED_OUT", +} +SUCCESS_CONCLUSIONS = {"NEUTRAL", "SKIPPED", "SUCCESS"} +PENDING_STATES = { + "EXPECTED", + "IN_PROGRESS", + "PENDING", + "QUEUED", + "REQUESTED", + "WAITING", +} +P0_TERMS = { + "critical", + "cve", + "data loss", + "priority:p0", + "release blocker", + "release-blocker", + "security", + "vulnerability", +} +P1_TERMS = { + "breaking", + "priority:p1", + "regression", + "sev1", + "sev2", +} +P3_TERMS = { + "chore", + "documentation", + "docs", + "example", + "examples", + "tests", +} +PR_FIELDS = ( + "number,title,url,state,author,isDraft,body,createdAt,additions,deletions," + "changedFiles,files,labels,comments,commits,reviewDecision,mergeable," + "mergeStateStatus,statusCheckRollup,closingIssuesReferences" +) +ACTIONABLE_RECOMMENDATIONS = { + "escalate_protected_pr", + "propose_close", + "propose_draft", + "second_owner_notification", + "warn_owner", +} +# Marker stage each enforcement action writes. Idempotency is keyed on the +# stage alone: the marker text also carries the run's score and as_of, so an +# exact-string comparison would never match a prior run's marker and the same +# notification would be posted again on every run. +ACTION_STAGES = { + "warn_owner": "warning", + "propose_draft": "draft", + "second_owner_notification": "draft", + "propose_close": "closed", + "escalate_protected_pr": "escalation", +} + + +def _is_leap(year: int) -> bool: + return year % 4 == 0 and (year % 100 != 0 or year % 400 == 0) + + +def _date_ordinal(value: str) -> int: + match = re.match(r"^(\d{4})-(\d{2})-(\d{2})", value) + if match is None: + raise ValueError(f"invalid ISO date: {value!r}") + year, month, day = (int(part) for part in match.groups()) + month_lengths = ( + 31, + 28 + int(_is_leap(year)), + 31, + 30, + 31, + 30, + 31, + 31, + 30, + 31, + 30, + 31, + ) + if year < 1 or not 1 <= month <= 12 or not 1 <= day <= month_lengths[month - 1]: + raise ValueError(f"invalid calendar date: {value!r}") + prior_years = year - 1 + leap_days = prior_years // 4 - prior_years // 100 + prior_years // 400 + return prior_years * 365 + leap_days + sum(month_lengths[: month - 1]) + day + + +def _validate_as_of(value: str) -> None: + if re.fullmatch(r"\d{4}-\d{2}-\d{2}", value) is None: + raise ValueError("as_of must be a calendar date in YYYY-MM-DD form") + _date_ordinal(value) + + +def _days_between(earlier: str, later: str) -> int: + return max(0, _date_ordinal(later) - _date_ordinal(earlier)) + + +def _repo_storage_key(repo: str) -> str: + return quote(repo, safe="") + + +def _run_gh(args: list[str]) -> Any: + completed = subprocess.run( + ["gh", *args], + check=False, + capture_output=True, + text=True, + ) + if completed.returncode != 0: + detail = completed.stderr.strip() or completed.stdout.strip() or "unknown gh error" + raise RuntimeError(f"gh command failed ({completed.returncode}): {detail}") + try: + return json.loads(completed.stdout) + except json.JSONDecodeError as exc: + raise RuntimeError("gh command did not return valid JSON") from exc + + +def _run_gh_command(args: list[str]) -> str: + completed = subprocess.run( + ["gh", *args], + check=False, + capture_output=True, + text=True, + ) + if completed.returncode != 0: + detail = completed.stderr.strip() or completed.stdout.strip() or "unknown gh error" + raise RuntimeError(f"gh command failed ({completed.returncode}): {detail}") + return completed.stdout.strip() + + +def _write_json(path: Path, value: Any) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + temporary = path.with_suffix(f"{path.suffix}.tmp") + temporary.write_text( + f"{json.dumps(value, indent=2, sort_keys=True)}\n", + encoding="utf-8", + ) + temporary.replace(path) + + +def _read_object(path: Path) -> dict[str, Any]: + value = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(value, dict): + raise ValueError(f"{path} must contain a JSON object") + return value + + +def _login(value: Any) -> str: + if isinstance(value, dict): + login = value.get("login") + if isinstance(login, str): + return login + return "" + + +def _labels(pr: dict[str, Any]) -> list[str]: + result = [] + for label in pr.get("labels") or []: + if isinstance(label, dict) and isinstance(label.get("name"), str): + result.append(label["name"].strip().lower()) + return sorted(set(result)) + + +def _latest_commit_at(pr: dict[str, Any]) -> str: + """Newest commit date on the PR, falling back to its creation date. + + ``gh pr view --json commits`` may return a truncated page on PRs with very + long histories. A missed newer commit only makes the PR look more idle than + it is, which lowers its score; the live re-score before any mutation reads + the same field, so enforcement never acts on a stale page alone. + """ + dates = [] + for commit in pr.get("commits") or []: + if not isinstance(commit, dict): + continue + value = commit.get("committedDate") or commit.get("authoredDate") + if isinstance(value, str): + dates.append(value) + if dates: + return max(dates) + created = pr.get("createdAt") + if not isinstance(created, str): + raise ValueError(f"PR #{pr.get('number')} has no usable activity date") + return created + + +def _ci_component(pr: dict[str, Any]) -> tuple[int, str]: + checks = pr.get("statusCheckRollup") or [] + if not checks: + return 5, "missing" + + has_pending = False + has_failure = False + for check in checks: + if not isinstance(check, dict): + continue + conclusion = str(check.get("conclusion") or "").upper() + state = str(check.get("state") or check.get("status") or "").upper() + if conclusion in FAILURE_CONCLUSIONS or state in FAILURE_CONCLUSIONS: + has_failure = True + elif state in PENDING_STATES or not conclusion and state not in SUCCESS_CONCLUSIONS: + has_pending = True + elif conclusion and conclusion not in SUCCESS_CONCLUSIONS: + has_failure = True + + if has_failure: + return 0, "failing" + if has_pending: + return 12, "pending" + return 20, "passing" + + +def _merge_component(pr: dict[str, Any]) -> tuple[int, str]: + mergeable = str(pr.get("mergeable") or "UNKNOWN").upper() + state = str(pr.get("mergeStateStatus") or "UNKNOWN").upper() + if mergeable == "CONFLICTING" or state == "DIRTY": + return 0, "conflicting" + if mergeable == "UNKNOWN" or state == "UNKNOWN": + return 5, "unknown" + if state == "CLEAN": + return 15, "clean" + return 10, state.lower() + + +def _review_component(pr: dict[str, Any]) -> tuple[int, str]: + if bool(pr.get("isDraft")): + return 5, "draft" + decision = str(pr.get("reviewDecision") or "REVIEW_REQUIRED").upper() + if decision == "APPROVED": + return 15, "approved" + if decision == "CHANGES_REQUESTED": + return 0, "changes_requested" + return 10, "review_required" + + +def _engagement_component(idle_days: int) -> int: + if idle_days <= 3: + return 40 + if idle_days <= 7: + return 32 + if idle_days <= 14: + return 24 + if idle_days <= 21: + return 16 + if idle_days <= 30: + return 8 + return 0 + + +def _completeness_component(pr: dict[str, Any]) -> tuple[int, dict[str, int]]: + body = str(pr.get("body") or "").strip() + files = pr.get("files") or [] + paths = [str(item.get("path") or "") for item in files if isinstance(item, dict)] + description = 3 if len(body) >= 200 else 0 + linked_issue = bool(pr.get("closingIssuesReferences")) or bool(ISSUE_REF_RE.search(body)) + rationale = 2 if linked_issue or RATIONALE_RE.search(body) else 0 + has_test_file = any( + path.startswith(("test/", "tests/", "web/src/test/")) + or "/test_" in path + or path.endswith((".spec.ts", ".spec.tsx", ".test.ts", ".test.tsx")) + for path in paths + ) + tests = 3 if has_test_file or TEST_RE.search(body) else 0 + changed_files = int(pr.get("changedFiles") or len(paths)) + churn = int(pr.get("additions") or 0) + int(pr.get("deletions") or 0) + if changed_files <= 25 and churn <= 1000: + focus = 2 + elif changed_files <= 50 and churn <= 3000: + focus = 1 + else: + focus = 0 + details = { + "description": description, + "rationale": rationale, + "tests": tests, + "focus": focus, + } + return sum(details.values()), details + + +def _priority(pr: dict[str, Any]) -> tuple[str, list[str]]: + labels = _labels(pr) + title = str(pr.get("title") or "").lower() + evidence = sorted(set(labels + [title])) + joined = " ".join(evidence) + p0_matches = sorted(term for term in P0_TERMS if term in joined) + if p0_matches: + return "P0", p0_matches + p1_matches = sorted(term for term in P1_TERMS if term in joined) + if p1_matches: + return "P1", p1_matches + if labels and all(any(term in label for term in P3_TERMS) for label in labels): + return "P3", labels + if any(term in title for term in P3_TERMS) and not any( + label in {"bug", "feature", "enhancement"} for label in labels + ): + return "P3", ["title"] + return "P2", labels or ["default"] + + +def _next_actor( + pr: dict[str, Any], + ci_status: str, + merge_status: str, + review_status: str, +) -> str: + if bool(pr.get("isDraft")): + return "OWNER" + if merge_status == "conflicting" or review_status == "changes_requested": + return "OWNER" + if ci_status == "failing": + return "OWNER" + if ci_status == "pending": + return "CI" + return "MAINTAINER" + + +def _category(score: int) -> str: + if score >= 85: + return "healthy" + if score >= 70: + return "active" + if score >= 60: + return "watch" + if score >= 51: + return "at_risk" + if score >= 30: + return "stalled" + return "abandoned" + + +def _lifecycle_marker(pr: dict[str, Any]) -> dict[str, Any] | None: + """Return the marker that owns the PR's current grace period. + + The furthest-advanced stage wins, and within that stage the earliest + marker wins. Selecting by recency instead would let a later warning-stage + comment shadow an existing draft marker, so the draft grace period would + restart on every run and the closure branch would never be re-evaluated. + Lifecycle progression therefore does not depend on the run cadence. + + Only comments authored by the authenticated identity count, and ``gh pr + view`` may return a truncated comment page on very busy PRs; a marker that + falls outside that page reads as absent, which is the conservative + direction (the ladder restarts rather than escalating). + """ + markers = [] + for comment in pr.get("comments") or []: + if not isinstance(comment, dict): + continue + if comment.get("viewerDidAuthor") is not True: + continue + body = str(comment.get("body") or "") + created_at = comment.get("createdAt") + if not isinstance(created_at, str): + continue + for match in MARKER_RE.finditer(body): + markers.append( + { + "stage": match.group(1), + "score": int(match.group(2)), + "as_of": match.group(3), + "created_at": created_at, + } + ) + if not markers: + return None + return min( + markers, + key=lambda item: (-STAGE_RANK[item["stage"]], item["created_at"]), + ) + + +def _owner_responded_after(pr: dict[str, Any], marker_at: str) -> bool: + owner = _login(pr.get("author")) + if _latest_commit_at(pr) > marker_at: + return True + for comment in pr.get("comments") or []: + if not isinstance(comment, dict): + continue + if _login(comment.get("author")) != owner: + continue + created_at = comment.get("createdAt") + if isinstance(created_at, str) and created_at > marker_at: + return True + return False + + +def _observation_streak( + previous: dict[str, Any] | None, + as_of: str, + score: int, + next_actor: str, +) -> int: + qualifies = score < 60 and next_actor == "OWNER" + if not qualifies: + return 0 + if not previous: + return 1 + previous_as_of = previous.get("last_as_of") + if not isinstance(previous_as_of, str): + return 1 + try: + stale = _date_ordinal(as_of) < _date_ordinal(previous_as_of) + except ValueError: + # Unparsable persisted date (hand-edited or written by a future + # schema): restart this PR's streak rather than aborting the run. + return 1 + if stale: + # Backfill, clock skew, or a reopened PR carrying old state. Restarting + # this PR's streak is the conservative direction — it delays the first + # warning by one observation instead of aborting scoring for every + # other PR in the run. + return 1 + if as_of == previous_as_of: + return int(previous.get("below60_owner_streak") or 1) + prior_qualified = ( + int(previous.get("last_score") or 100) < 60 and previous.get("last_next_actor") == "OWNER" + ) + return int(previous.get("below60_owner_streak") or 0) + 1 if prior_qualified else 1 + + +def _recommend_action( + pr: dict[str, Any], + raw_score: int, + priority: str, + next_actor: str, + marker: dict[str, Any] | None, + streak: int, + as_of: str, +) -> tuple[int, str, list[str]]: + score = raw_score + reasons = [] + protected = ( + priority in {"P0", "P1"} or str(pr.get("reviewDecision") or "").upper() == "APPROVED" + ) + + if marker: + marker_age = _days_between(marker["created_at"], as_of) + responded = _owner_responded_after(pr, marker["created_at"]) + reasons.append(f"{marker['stage']}_marker_age={marker_age}") + if responded: + reasons.append("owner_responded") + return score, "monitor_response", reasons + if marker["stage"] == "draft" and marker_age >= 14: + score = max(0, score - 25) + reasons.append("ignored_intervention=-25") + if score < 30: + if protected: + return score, "escalate_protected_pr", reasons + return score, "propose_close", reasons + if marker["stage"] == "warning" and marker_age >= 7 and score <= 50: + if bool(pr.get("isDraft")): + return score, "second_owner_notification", reasons + if protected: + return score, "escalate_protected_pr", reasons + return score, "propose_draft", reasons + if marker["stage"] in {"warning", "draft"} and score < 60: + # The stage's notification is already on the PR and its grace period + # has not expired. Falling through to the score<60 ladder would + # re-recommend warn_owner — a stage this PR has already passed — so + # hold here instead of walking the lifecycle backwards. + reasons.append(f"awaiting_{marker['stage']}_grace_period") + return score, "await_owner_deadline", reasons + # stage=closed means a previous close was reverted (PR reopened); the + # ladder restarts from observation below. + + if score < 60: + if next_actor == "CI": + return score, "await_ci", reasons + if next_actor != "OWNER": + return score, "alert_maintainers", reasons + if protected: + return score, "escalate_protected_pr", reasons + if streak >= 2: + return score, "warn_owner", reasons + return score, "observe_again", reasons + return score, "none", reasons + + +def _score_pr( + pr: dict[str, Any], + as_of: str, + previous: dict[str, Any] | None, +) -> tuple[dict[str, Any], dict[str, Any]]: + number = pr.get("number") + if isinstance(number, bool) or not isinstance(number, int): + raise ValueError("PR number must be an integer") + last_commit_at = _latest_commit_at(pr) + idle_days = _days_between(last_commit_at, as_of) + ci_points, ci_status = _ci_component(pr) + merge_points, merge_status = _merge_component(pr) + review_points, review_status = _review_component(pr) + engagement_points = _engagement_component(idle_days) + completeness_points, completeness_details = _completeness_component(pr) + raw_score = ci_points + merge_points + review_points + engagement_points + completeness_points + priority, priority_evidence = _priority(pr) + next_actor = _next_actor(pr, ci_status, merge_status, review_status) + marker = _lifecycle_marker(pr) + streak = _observation_streak(previous, as_of, raw_score, next_actor) + score, action, action_reasons = _recommend_action( + pr, + raw_score, + priority, + next_actor, + marker, + streak, + as_of, + ) + result = { + "number": number, + "title": str(pr.get("title") or ""), + "url": str(pr.get("url") or ""), + "owner": _login(pr.get("author")), + "is_draft": bool(pr.get("isDraft")), + "last_commit_at": last_commit_at, + "idle_days": idle_days, + "score": score, + "raw_score": raw_score, + "category": _category(score), + "priority": priority, + "priority_evidence": priority_evidence, + "next_actor": next_actor, + "recommended_action": action, + "action_reasons": action_reasons, + "below60_owner_streak": streak, + "lifecycle_marker": marker, + "components": { + "ci": {"points": ci_points, "status": ci_status}, + "mergeability": {"points": merge_points, "status": merge_status}, + "review": {"points": review_points, "status": review_status}, + "engagement": {"points": engagement_points, "idle_days": idle_days}, + "completeness": { + "points": completeness_points, + **completeness_details, + }, + "ignored_intervention": score - raw_score, + }, + } + next_state = { + "last_as_of": as_of, + "last_score": score, + "last_next_actor": next_actor, + "below60_owner_streak": streak, + } + return result, next_state + + +def _fetch_snapshot(repo: str, as_of: str, snapshot_id: str, max_prs: int) -> dict[str, Any]: + rows = _run_gh( + [ + "pr", + "list", + "--repo", + repo, + "--state", + "open", + "--limit", + str(max_prs + 1), + "--json", + "number", + ] + ) + if not isinstance(rows, list): + raise RuntimeError("gh pr list did not return a JSON list") + if len(rows) > max_prs: + raise RuntimeError(f"open PR count exceeds max_prs={max_prs}") + + prs = [] + numbers = sorted(int(row["number"]) for row in rows) + for number in numbers: + value = _run_gh( + [ + "pr", + "view", + str(number), + "--repo", + repo, + "--json", + PR_FIELDS, + ] + ) + if not isinstance(value, dict): + raise RuntimeError(f"gh pr view {number} did not return an object") + prs.append(value) + return { + "schema_version": SCHEMA_VERSION, + "repo": repo, + "as_of": as_of, + "snapshot_id": snapshot_id, + "prs": prs, + } + + +def _render_report(repo: str, as_of: str, scores: list[dict[str, Any]], mode: str) -> str: + lines = [ + "# PR Health Report", + "", + f"- Repository: `{repo}`", + f"- As of: `{as_of}`", + f"- Open PRs: {len(scores)}", + "- Scoring: deterministic schema v1", + "", + "| PR | Score | Health | Priority | Idle | Next actor | Recommendation |", + "|---:|---:|---|---|---:|---|---|", + ] + for item in sorted(scores, key=lambda value: (value["score"], value["number"])): + lines.append( + "| " + f"[#{item['number']}]({item['url']}) | {item['score']} | " + f"{item['category']} | {item['priority']} | {item['idle_days']}d | " + f"{item['next_actor']} | {item['recommended_action']} |" + ) + lines.extend( + [ + "", + "## Score Rules", + "", + "- CI: 20 passing, 12 pending, 5 missing, 0 failing.", + "- Mergeability: 15 clean, 10 blocked/behind, 5 unknown, 0 conflicting.", + "- Review: 15 approved, 10 review required, 5 draft, 0 changes requested.", + "- Engagement: 40/32/24/16/8/0 across <=3/7/14/21/30/>30 idle days.", + "- Completeness: 3 description, 2 rationale, 3 tests, 2 focused scope.", + "- Ignoring a draft-stage notification for 14 days applies -25.", + "", + ( + "Dry-run mode: this report does not mutate GitHub." + if mode == "dry_run" + else "Apply mode: eligible recommendations are live-revalidated before enforcement." + ), + "", + ] + ) + return "\n".join(lines) + + +def _importance_prompt(report_path: Path, scores_path: Path, repo: str, as_of: str) -> str: + return f"""Review the deterministic PR health artifacts for {repo} as of {as_of}. + +Read: +- {report_path} +- {scores_path} + +Produce a concise maintainer synthesis grouped by: +1. protected P0/P1 PRs needing escalation or adoption, +2. owner-blocked PRs needing attention, +3. maintainer-blocked PRs, +4. closure candidates and the evidence supporting them. + +The score, category, next_actor, priority, and recommended_action fields are +authoritative rule outputs. Do not recalculate, override, or invent scores. +Call out uncertain importance classifications as advisory. Return Markdown only. +Do not modify files or GitHub.""" + + +def _action_marker(action: str, score: int, as_of: str) -> str: + if action == "warn_owner": + return f"" + if action in {"propose_draft", "second_owner_notification"}: + return f"" + if action == "propose_close": + return f"" + if action == "escalate_protected_pr": + return f"" + raise ValueError(f"unsupported enforcement action: {action}") + + +def _blocker_lines(item: dict[str, Any]) -> list[str]: + components = item["components"] + return [ + f"- CI: {components['ci']['status']}", + f"- Mergeability: {components['mergeability']['status']}", + f"- Review: {components['review']['status']}", + f"- Last commit: {item['idle_days']} days ago", + ] + + +def _comment_body(item: dict[str, Any], as_of: str) -> str: + action = item["recommended_action"] + owner = item["owner"] + score = item["score"] + blockers = "\n".join(_blocker_lines(item)) + marker = _action_marker(action, score, as_of) + + if action == "warn_owner": + message = f"""@{owner} This PR's automated health score is **{score}/100** and needs attention. + +Current signals: +{blockers} + +No state change is being made now. Please push an update or reply with your plan within 7 days. The score will be recalculated before any further action.""" + elif action == "propose_draft": + message = f"""@{owner} This PR's automated health score is now **{score}/100**. The previous notification has been open for at least 7 days without new activity. + +Current signals: +{blockers} + +The PR is being moved to draft while the outstanding issues are addressed. Please push an update or reply with a concrete plan within 14 days. No closure will occur without another evaluation.""" + elif action == "second_owner_notification": + message = f"""@{owner} This draft PR's automated health score is **{score}/100**. The previous notification has been open for at least 7 days without new activity. + +Current signals: +{blockers} + +Please push an update or reply with a concrete plan within 14 days. No closure will occur without another evaluation.""" + elif action == "propose_close": + message = f"""@{owner} This PR's automated health score is **{score}/100** after the warning and draft grace periods. + +Current signals: +{blockers} + +The PR remains blocked and no owner activity was detected, so it is being closed to keep the active backlog current. This is not a rejection of the proposal. It can be reopened or resubmitted when the work is ready to continue.""" + elif action == "escalate_protected_pr": + protections = [] + if item["priority"] in {"P0", "P1"}: + protections.append(f"priority {item['priority']}") + if item["components"]["review"]["status"] == "approved": + protections.append("approved review state") + protection_text = " and ".join(protections) or "protected status" + message = f"""@{owner} This PR's automated health score is **{score}/100** and requires attention. + +Current signals: +{blockers} + +This PR is protected from automated draft or closure because of its {protection_text}. Maintainer escalation is requested. Please push an update or reply with the intended next step.""" + else: + raise ValueError(f"unsupported enforcement action: {action}") + + return f"{message}\n\n{marker}" + + +def _authored_bodies(pr: dict[str, Any]) -> list[str]: + return [ + str(comment.get("body") or "") + for comment in pr.get("comments") or [] + if isinstance(comment, dict) and comment.get("viewerDidAuthor") is True + ] + + +def _has_marker_for_stage(pr: dict[str, Any], stage: str) -> bool: + """Has this workflow identity already posted a marker for ``stage``? + + Dedupe is by stage presence, never by exact marker text: markers embed the + run's ``score`` and ``as_of``, so exact matching would re-post the same + notification every run. + """ + for body in _authored_bodies(pr): + if stage == "escalation": + if ESCALATION_MARKER_RE.search(body): + return True + continue + if any(match.group(1) == stage for match in MARKER_RE.finditer(body)): + return True + return False + + +def _has_marker_for_action(pr: dict[str, Any], action: str) -> bool: + stage = ACTION_STAGES.get(action) + if stage is None: + raise ValueError(f"unsupported enforcement action: {action}") + return _has_marker_for_stage(pr, stage) + + +def _fetch_pr(repo: str, number: int) -> dict[str, Any]: + value = _run_gh( + [ + "pr", + "view", + str(number), + "--repo", + repo, + "--json", + PR_FIELDS, + ] + ) + if not isinstance(value, dict): + raise RuntimeError(f"gh pr view {number} did not return an object") + return value + + +def _parse_close_allowlist(value: str) -> set[int]: + if not value.strip(): + return set() + result = set() + for token in value.split(","): + token = token.strip() + if not token.isdigit() or int(token) < 1: + raise ValueError("close_allowlist must be a comma-separated list of PR numbers") + result.add(int(token)) + return result + + +def _apply_recommendations( + repo: str, + as_of: str, + scores: list[dict[str, Any]], + persisted_state: dict[str, Any], + close_allowlist: set[int], + journal_path: Path, +) -> list[dict[str, Any]]: + results = [] + state_by_pr = persisted_state.get("prs") or {} + + for planned in sorted(scores, key=lambda item: int(item["number"])): + action = str(planned["recommended_action"]) + if action not in ACTIONABLE_RECOMMENDATIONS: + continue + number = int(planned["number"]) + result: dict[str, Any] = { + "number": number, + "planned_action": action, + "status": "pending", + } + try: + live_pr = _fetch_pr(repo, number) + if live_pr.get("state") != "OPEN": + result["status"] = "skipped_not_open" + results.append(result) + _write_json( + journal_path, + { + "schema_version": SCHEMA_VERSION, + "repo": repo, + "as_of": as_of, + "results": results, + }, + ) + continue + if _has_marker_for_action(live_pr, action): + result["status"] = "already_applied" + results.append(result) + _write_json( + journal_path, + { + "schema_version": SCHEMA_VERSION, + "repo": repo, + "as_of": as_of, + "results": results, + }, + ) + continue + live_score, _ = _score_pr( + live_pr, + as_of, + state_by_pr.get(str(number)), + ) + if ( + live_score["score"] != planned["score"] + or live_score["recommended_action"] != action + ): + result.update( + { + "status": "skipped_live_drift", + "live_score": live_score["score"], + "live_recommendation": live_score["recommended_action"], + } + ) + elif action == "propose_close" and number not in close_allowlist: + result["status"] = "skipped_closure_not_allowlisted" + else: + body = _comment_body(planned, as_of) + if action == "propose_draft": + # propose_draft is only recommended for a non-draft PR, and + # the live re-score above rejects any drift, so the PR is + # known not to be a draft here. + _run_gh_command(["pr", "ready", str(number), "--repo", repo, "--undo"]) + _run_gh_command(["pr", "comment", str(number), "--repo", repo, "--body", body]) + result["status"] = "drafted_and_commented" + elif action == "propose_close": + _run_gh_command( + [ + "pr", + "close", + str(number), + "--repo", + repo, + "--comment", + body, + ] + ) + result["status"] = "closed_and_commented" + else: + _run_gh_command(["pr", "comment", str(number), "--repo", repo, "--body", body]) + result["status"] = "commented" + except Exception as exc: + result.update({"status": "error", "error": str(exc)}) + results.append(result) + _write_json( + journal_path, + { + "schema_version": SCHEMA_VERSION, + "repo": repo, + "as_of": as_of, + "results": results, + }, + ) + return results + + +def _run_locked(inputs: dict[str, Any]) -> None: + repo = str(inputs.get("repo") or "").strip() + as_of = str(inputs.get("as_of") or "").strip() + snapshot_id = str(inputs.get("snapshot_id") or "").strip() + max_prs = inputs.get("max_prs", 500) + importance_analysis = inputs.get("importance_analysis", True) + importance_provider = str(inputs.get("importance_provider") or "").strip() + importance_agent = str(inputs.get("importance_agent") or "").strip() + mode = str(inputs.get("mode") or "").strip() + close_allowlist_text = str(inputs.get("close_allowlist") or "").strip() + + if repo.count("/") != 1 or any(not part for part in repo.split("/")): + raise ValueError("repo must be in owner/name form") + _validate_as_of(as_of) + if not SAFE_ID_RE.fullmatch(snapshot_id) or snapshot_id in RESERVED_IDS: + raise ValueError( + "snapshot_id may contain only letters, numbers, dot, underscore, and dash, " + "and may not be '.' or '..'" + ) + if isinstance(max_prs, bool) or not isinstance(max_prs, int) or not 1 <= max_prs <= 2000: + raise ValueError("max_prs must be an integer from 1 through 2000") + if not isinstance(importance_analysis, bool): + raise ValueError("importance_analysis must be boolean") + if not SAFE_ID_RE.fullmatch(importance_provider): + raise ValueError("importance_provider must be a nonempty safe identifier") + if not SAFE_ID_RE.fullmatch(importance_agent): + raise ValueError("importance_agent must be a nonempty safe identifier") + if mode not in {"dry_run", "apply"}: + raise ValueError("mode must be dry_run or apply") + close_allowlist = _parse_close_allowlist(close_allowlist_text) + if mode == "dry_run" and close_allowlist: + raise ValueError("close_allowlist is only valid in apply mode") + + repo_key = _repo_storage_key(repo) + root = Path.home() / ".local" / "state" / "cao" / "pr-health" / repo_key + artifact_dir = root / "runs" / snapshot_id + snapshot_path = artifact_dir / "snapshot.json" + scores_path = artifact_dir / "scores.json" + report_path = artifact_dir / "report.md" + analysis_path = artifact_dir / "importance-analysis.md" + enforcement_path = artifact_dir / "enforcement.json" + manifest_path = artifact_dir / "manifest.json" + state_path = root / "state.json" + + if manifest_path.is_file(): + manifest = _read_object(manifest_path) + if ( + manifest.get("repo") != repo + or manifest.get("as_of") != as_of + or manifest.get("snapshot_id") != snapshot_id + or manifest.get("mode") != mode + ): + raise ValueError("snapshot_id already exists with different inputs") + emit_output(manifest) + return + + artifact_dir.mkdir(parents=True, exist_ok=True) + if snapshot_path.is_file(): + snapshot = _read_object(snapshot_path) + if ( + snapshot.get("repo") != repo + or snapshot.get("as_of") != as_of + or snapshot.get("snapshot_id") != snapshot_id + ): + raise ValueError("existing snapshot does not match requested inputs") + else: + snapshot = _fetch_snapshot(repo, as_of, snapshot_id, max_prs) + _write_json(snapshot_path, snapshot) + + state = ( + _read_object(state_path) + if state_path.is_file() + else {"schema_version": SCHEMA_VERSION, "repo": repo, "prs": {}} + ) + if ( + state.get("schema_version") != SCHEMA_VERSION + or state.get("repo") != repo + or not isinstance(state.get("prs"), dict) + ): + raise ValueError("persisted PR health state has an unsupported schema") + + scores = [] + next_pr_state = dict(state["prs"]) + for pr in sorted(snapshot.get("prs") or [], key=lambda value: int(value["number"])): + number_key = str(pr["number"]) + previous = state["prs"].get(number_key) + if previous is not None and not isinstance(previous, dict): + raise ValueError(f"persisted state for PR #{number_key} is invalid") + score, pr_state = _score_pr(pr, as_of, previous) + scores.append(score) + next_pr_state[number_key] = pr_state + + open_numbers = {str(item["number"]) for item in scores} + next_pr_state = { + number: value for number, value in next_pr_state.items() if number in open_numbers + } + updated_state = { + "schema_version": SCHEMA_VERSION, + "repo": repo, + "prs": next_pr_state, + } + _write_json(state_path, updated_state) + _write_json( + scores_path, + { + "schema_version": SCHEMA_VERSION, + "repo": repo, + "as_of": as_of, + "snapshot_id": snapshot_id, + "scores": scores, + }, + ) + report_path.write_text( + _render_report(repo, as_of, scores, mode), + encoding="utf-8", + ) + + analysis_error = None + if importance_analysis: + try: + handle = run_step( + importance_provider, + importance_agent, + _importance_prompt(report_path, scores_path, repo, as_of), + step_id=f"importance-{snapshot_id}", + timeout=1800.0, + ) + analysis_path.write_text(f"{(handle.output or '').strip()}\n", encoding="utf-8") + except ShimError as exc: + analysis_error = str(exc) + + enforcement_results = [] + if mode == "apply": + enforcement_results = _apply_recommendations( + repo, + as_of, + scores, + updated_state, + close_allowlist, + enforcement_path, + ) + + actions: dict[str, int] = {} + for item in scores: + actions[item["recommended_action"]] = actions.get(item["recommended_action"], 0) + 1 + manifest = { + "schema_version": SCHEMA_VERSION, + "repo": repo, + "as_of": as_of, + "snapshot_id": snapshot_id, + "mode": mode, + "open_prs": len(scores), + "actions": dict(sorted(actions.items())), + "snapshot_file": str(snapshot_path), + "scores_file": str(scores_path), + "report_file": str(report_path), + "importance_analysis_file": (str(analysis_path) if analysis_path.is_file() else None), + "importance_analysis_error": analysis_error, + "enforcement_file": (str(enforcement_path) if enforcement_path.is_file() else None), + "enforcement_results": enforcement_results, + "mutated_github": any( + result.get("status") + in { + "closed_and_commented", + "commented", + "drafted_and_commented", + } + for result in enforcement_results + ), + } + _write_json(manifest_path, manifest) + emit_output(manifest) + + +def _acquire_repo_lock(repo: str) -> tuple[TextIO, Path]: + root = Path.home() / ".local" / "state" / "cao" / "pr-health" / _repo_storage_key(repo) + root.mkdir(parents=True, exist_ok=True) + lock_path = root / ".workflow.lock" + lock_handle = lock_path.open("a+", encoding="utf-8") + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX) + return lock_handle, lock_path + + +def main() -> None: + inputs = get_inputs() + repo = str(inputs.get("repo") or "").strip() + if repo.count("/") != 1 or any(not part for part in repo.split("/")): + raise ValueError("repo must be in owner/name form") + + lock_handle, _ = _acquire_repo_lock(repo) + try: + _run_locked(inputs) + finally: + fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN) + lock_handle.close() + + +if __name__ == "__main__": + main() diff --git a/examples/workflows/pr-health/pr_health_biweekly_guard.py b/examples/workflows/pr-health/pr_health_biweekly_guard.py new file mode 100755 index 000000000..c05dc09d6 --- /dev/null +++ b/examples/workflows/pr-health/pr_health_biweekly_guard.py @@ -0,0 +1,55 @@ +#!/usr/bin/env python3 +"""Gate a weekly CAO flow to an exact 14-day cadence.""" + +from __future__ import annotations + +import json +from datetime import date, datetime, timezone + +# Customize these two values before registering the flow. +REPOSITORY = "awslabs/cli-agent-orchestrator" +ANCHOR = date(2026, 1, 5) # Monday; due dates repeat every 14 days. + + +def is_due(today: date) -> bool: + days = (today - ANCHOR).days + return days >= 0 and days % 14 == 0 + + +def today_utc() -> date: + """Today in UTC. + + The workflow's idle/threshold math compares ``as_of`` against GitHub's UTC + timestamps, so deriving ``as_of`` from the server's local date would shift + the 7/14/21-day threshold crossings by a day for runs near midnight. + """ + return datetime.now(timezone.utc).date() + + +def main() -> None: + today = today_utc() + as_of = today.isoformat() + due = is_due(today) + # Identifiers are mode-qualified so the dry-run and apply flows can both be + # registered: a shared snapshot_id would make the second flow to run on a + # due date collide with the first flow's manifest. + print( + json.dumps( + { + "execute": due, + "output": { + "as_of": as_of, + "repo": REPOSITORY, + "run_id_dry_run": f"pr-health-biweekly-dry-run-{as_of}", + "snapshot_id_dry_run": f"scheduled-dry-run-{as_of}", + "run_id_apply": f"pr-health-biweekly-apply-{as_of}", + "snapshot_id_apply": f"scheduled-apply-{as_of}", + }, + }, + separators=(",", ":"), + ) + ) + + +if __name__ == "__main__": + main() diff --git a/test/examples/test_pr_health_workflow_example.py b/test/examples/test_pr_health_workflow_example.py new file mode 100644 index 000000000..d166ea7fc --- /dev/null +++ b/test/examples/test_pr_health_workflow_example.py @@ -0,0 +1,1184 @@ +"""Tests for the deterministic PR-health workflow example. + +The example lives under ``examples/`` and is loaded by path, so it is not +covered by ``--cov=src``. These tests are therefore the only coverage the +scoring engine, the marker lifecycle, and the GitHub-mutating enforcement +branches get; the workflow itself carries no in-product assert bundle. +""" + +from __future__ import annotations + +import importlib.util +from datetime import date, datetime, timezone +from pathlib import Path +from types import ModuleType +from typing import Any + +import pytest + +from cli_agent_orchestrator.services.flow_service import _parse_flow_file +from cli_agent_orchestrator.services.script_lint import lint_script + +REPO_ROOT = Path(__file__).resolve().parents[2] +EXAMPLE_DIR = REPO_ROOT / "examples" / "workflows" / "pr-health" + + +def _load_module(name: str, path: Path) -> ModuleType: + spec = importlib.util.spec_from_file_location(name, path) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +@pytest.fixture(scope="module") +def workflow() -> ModuleType: + return _load_module("pr_health_example", EXAMPLE_DIR / "pr_health.py") + + +@pytest.fixture(scope="module") +def guard() -> ModuleType: + return _load_module( + "pr_health_biweekly_guard", + EXAMPLE_DIR / "pr_health_biweekly_guard.py", + ) + + +@pytest.fixture +def base_pr() -> dict[str, Any]: + """A maximally healthy PR: every component at full points, score 100.""" + return { + "number": 1, + "title": "feat: deterministic workflow", + "url": "https://example.invalid/pull/1", + "author": {"login": "owner"}, + "isDraft": False, + "body": ( + "This change fixes #1 because the workflow needs deterministic scoring. " + "Testing and verification cover every score boundary. " * 3 + ), + "createdAt": "2026-07-01T00:00:00Z", + "additions": 100, + "deletions": 20, + "changedFiles": 4, + "files": [{"path": "test/test_pr_health.py"}], + "labels": [{"name": "feature"}], + "comments": [], + "commits": [{"committedDate": "2026-07-31T00:00:00Z"}], + "reviewDecision": "APPROVED", + "mergeable": "MERGEABLE", + "mergeStateStatus": "CLEAN", + "statusCheckRollup": [{"status": "COMPLETED", "conclusion": "SUCCESS"}], + "closingIssuesReferences": [{"number": 1}], + } + + +def _marker(stage: str, score: int, as_of: str) -> str: + return f"" + + +def _authored_comment(body: str, created_at: str) -> dict[str, Any]: + return { + "author": {"login": "maintainer"}, + "createdAt": created_at, + "body": body, + "viewerDidAuthor": True, + } + + +@pytest.fixture +def at_risk_pr(base_pr: dict[str, Any]) -> dict[str, Any]: + """Score 56: passing CI, blocked merge, changes requested, 17 idle days.""" + return { + **base_pr, + "reviewDecision": "CHANGES_REQUESTED", + "mergeStateStatus": "BLOCKED", + "commits": [{"committedDate": "2026-07-14T00:00:00Z"}], + } + + +@pytest.fixture +def warned_pr(at_risk_pr: dict[str, Any]) -> dict[str, Any]: + """Score 44 with a 7-day-old warning marker: draft-eligible.""" + return { + **at_risk_pr, + "statusCheckRollup": [{"status": "COMPLETED", "conclusion": "FAILURE"}], + "commits": [{"committedDate": "2026-07-23T00:00:00Z"}], + "comments": [ + _authored_comment(_marker("warning", 44, "2026-07-24"), "2026-07-24T00:00:00Z") + ], + } + + +# -------------------------------------------------------------------------- +# Static validation and wiring +# -------------------------------------------------------------------------- + + +def test_workflow_passes_static_validation() -> None: + path = EXAMPLE_DIR / "pr_health.py" + result = lint_script(path.read_text(encoding="utf-8"), str(path)) + + assert result.status == "pass" + assert result.findings == [] + + +def test_repo_storage_key_is_unambiguous(workflow: ModuleType) -> None: + assert workflow._repo_storage_key("a--b/c") != workflow._repo_storage_key("a/b--c") + + +# -------------------------------------------------------------------------- +# Calendar engine +# -------------------------------------------------------------------------- + + +def test_days_between_crosses_leap_and_non_leap_february(workflow: ModuleType) -> None: + assert workflow._days_between("2024-02-28", "2024-03-01") == 2 + assert workflow._days_between("2025-02-28", "2025-03-01") == 1 + + +@pytest.mark.parametrize( + ("earlier", "later", "expected"), + [ + ("2026-01-31", "2026-02-01", 1), + ("2026-12-31", "2027-01-01", 1), + ("2026-01-01", "2027-01-01", 365), + ("2024-01-01", "2025-01-01", 366), + ("2024-02-29", "2024-03-01", 1), + ("2026-07-31", "2026-07-31", 0), + ("2026-08-01", "2026-07-31", 0), # clamped, never negative + ("1900-02-28", "1900-03-01", 1), # 1900 is not a leap year + ("2000-02-28", "2000-03-01", 2), # 2000 is a leap year + ], +) +def test_days_between_boundaries( + workflow: ModuleType, + earlier: str, + later: str, + expected: int, +) -> None: + assert workflow._days_between(earlier, later) == expected + + +def test_date_ordinal_agrees_with_stdlib_across_a_dense_range(workflow: ModuleType) -> None: + """The bespoke ordinal must match ``date.toordinal`` offsets exactly.""" + samples = [ + date(year, month, day) + for year in (1900, 1999, 2000, 2024, 2025, 2026, 2100) + for month in range(1, 13) + for day in (1, 15, 28) + ] + for value in samples: + delta = workflow._date_ordinal(value.isoformat()) - value.toordinal() + assert delta == workflow._date_ordinal("2026-01-01") - date(2026, 1, 1).toordinal() + + +@pytest.mark.parametrize( + "value", + [ + "2026-02-30", + "2025-02-29", # not a leap year + "2026-13-01", + "2026-00-10", + "2026-01-00", + "2026-01-32", + "0000-01-01", + "2026-7-31", + "2026/07/31", + "2026-07-31T00:00:00Z", + "", + "not-a-date", + ], +) +def test_validate_as_of_rejects_invalid_dates(workflow: ModuleType, value: str) -> None: + with pytest.raises(ValueError): + workflow._validate_as_of(value) + + +def test_validate_as_of_accepts_a_leap_day(workflow: ModuleType) -> None: + workflow._validate_as_of("2024-02-29") + + +# -------------------------------------------------------------------------- +# Scoring +# -------------------------------------------------------------------------- + + +def test_engagement_bands(workflow: ModuleType) -> None: + days = (3, 4, 7, 8, 14, 15, 21, 22, 30, 31) + assert [workflow._engagement_component(day) for day in days] == [ + 40, + 32, + 32, + 24, + 24, + 16, + 16, + 8, + 8, + 0, + ] + + +def test_category_bands(workflow: ModuleType) -> None: + scores = (100, 85, 84, 70, 69, 60, 59, 51, 50, 30, 29) + assert [workflow._category(score) for score in scores] == [ + "healthy", + "healthy", + "active", + "active", + "watch", + "watch", + "at_risk", + "at_risk", + "stalled", + "stalled", + "abandoned", + ] + + +def test_healthy_pr_scores_100_and_recommends_nothing( + workflow: ModuleType, + base_pr: dict[str, Any], +) -> None: + healthy, _ = workflow._score_pr(base_pr, "2026-07-31", None) + + assert healthy["score"] == 100 + assert healthy["category"] == "healthy" + assert healthy["next_actor"] == "MAINTAINER" + assert healthy["recommended_action"] == "none" + + +def test_first_below_60_observation_only_observes( + workflow: ModuleType, + at_risk_pr: dict[str, Any], +) -> None: + at_risk, _ = workflow._score_pr(at_risk_pr, "2026-07-31", None) + + assert at_risk["score"] == 56 + assert at_risk["recommended_action"] == "observe_again" + assert at_risk["below60_owner_streak"] == 1 + + +def test_second_below_60_observation_warns_the_owner( + workflow: ModuleType, + at_risk_pr: dict[str, Any], +) -> None: + _, state = workflow._score_pr(at_risk_pr, "2026-07-31", None) + warning, _ = workflow._score_pr(at_risk_pr, "2026-08-01", state) + + assert warning["below60_owner_streak"] == 2 + assert warning["recommended_action"] == "warn_owner" + + +def test_unanswered_warning_after_seven_days_proposes_draft( + workflow: ModuleType, + warned_pr: dict[str, Any], +) -> None: + stalled, _ = workflow._score_pr(warned_pr, "2026-07-31", None) + + assert stalled["score"] == 44 + assert stalled["recommended_action"] == "propose_draft" + + +def test_ignored_draft_after_fourteen_days_proposes_close( + workflow: ModuleType, + base_pr: dict[str, Any], +) -> None: + abandoned_pr = { + **base_pr, + "isDraft": True, + "reviewDecision": "REVIEW_REQUIRED", + "commits": [{"committedDate": "2026-06-01T00:00:00Z"}], + "comments": [_authored_comment(_marker("draft", 50, "2026-07-17"), "2026-07-17T00:00:00Z")], + } + + abandoned, _ = workflow._score_pr(abandoned_pr, "2026-07-31", None) + + assert abandoned["raw_score"] == 50 + assert abandoned["score"] == 25 + assert abandoned["category"] == "abandoned" + assert abandoned["recommended_action"] == "propose_close" + + +def test_protected_pr_escalates_instead_of_closing( + workflow: ModuleType, + base_pr: dict[str, Any], +) -> None: + protected_pr = { + **base_pr, + "isDraft": True, + "reviewDecision": "REVIEW_REQUIRED", + "commits": [{"committedDate": "2026-06-01T00:00:00Z"}], + "comments": [_authored_comment(_marker("draft", 50, "2026-07-17"), "2026-07-17T00:00:00Z")], + "title": "fix(security): prevent command injection", + "labels": [{"name": "security"}], + } + + protected, _ = workflow._score_pr(protected_pr, "2026-07-31", None) + + assert protected["priority"] == "P0" + assert protected["score"] == 25 + assert protected["recommended_action"] == "escalate_protected_pr" + + +def test_forged_marker_from_another_author_is_ignored( + workflow: ModuleType, + warned_pr: dict[str, Any], +) -> None: + forged = { + **warned_pr, + "comments": [ + { + "author": {"login": "owner"}, + "createdAt": "2026-07-01T00:00:00Z", + "body": _marker("warning", 44, "2026-07-24"), + "viewerDidAuthor": False, + } + ], + } + + result, _ = workflow._score_pr(forged, "2026-07-31", None) + + assert result["lifecycle_marker"] is None + assert result["recommended_action"] == "observe_again" + + +# -------------------------------------------------------------------------- +# Observation streak: stale persisted state must not abort the run +# -------------------------------------------------------------------------- + + +def test_streak_restarts_instead_of_raising_on_stale_state(workflow: ModuleType) -> None: + """A backfill / reopened PR / clock skew must not kill the whole run.""" + previous = { + "last_as_of": "2026-08-15", + "last_score": 40, + "last_next_actor": "OWNER", + "below60_owner_streak": 3, + } + + assert workflow._observation_streak(previous, "2026-07-31", 40, "OWNER") == 1 + + +def test_streak_restarts_on_unparsable_persisted_date(workflow: ModuleType) -> None: + previous = {"last_as_of": "not-a-date", "below60_owner_streak": 9} + + assert workflow._observation_streak(previous, "2026-07-31", 40, "OWNER") == 1 + + +def test_stale_state_for_one_pr_does_not_stop_scoring_others( + workflow: ModuleType, + at_risk_pr: dict[str, Any], +) -> None: + stale = { + "last_as_of": "2026-12-01", + "last_score": 40, + "last_next_actor": "OWNER", + "below60_owner_streak": 5, + } + + first, _ = workflow._score_pr(at_risk_pr, "2026-07-31", stale) + second, _ = workflow._score_pr({**at_risk_pr, "number": 2}, "2026-07-31", None) + + assert first["below60_owner_streak"] == 1 + assert second["below60_owner_streak"] == 1 + + +def test_same_day_rerun_preserves_the_streak(workflow: ModuleType) -> None: + previous = { + "last_as_of": "2026-07-31", + "last_score": 40, + "last_next_actor": "OWNER", + "below60_owner_streak": 2, + } + + assert workflow._observation_streak(previous, "2026-07-31", 40, "OWNER") == 2 + + +def test_streak_resets_when_the_pr_recovered_in_between(workflow: ModuleType) -> None: + previous = { + "last_as_of": "2026-07-24", + "last_score": 90, + "last_next_actor": "MAINTAINER", + "below60_owner_streak": 0, + } + + assert workflow._observation_streak(previous, "2026-07-31", 40, "OWNER") == 1 + + +# -------------------------------------------------------------------------- +# Lifecycle marker selection: progression must not depend on run cadence +# -------------------------------------------------------------------------- + + +def test_draft_marker_wins_over_a_later_warning_marker( + workflow: ModuleType, + base_pr: dict[str, Any], +) -> None: + """A later warning comment must not shadow an existing draft marker. + + Selecting the newest marker instead would restart the draft grace period on + every run, so the closure branch would never be re-evaluated. + """ + pr = { + **base_pr, + "comments": [ + _authored_comment(_marker("draft", 50, "2026-07-01"), "2026-07-01T00:00:00Z"), + _authored_comment(_marker("warning", 45, "2026-07-20"), "2026-07-20T00:00:00Z"), + ], + } + + marker = workflow._lifecycle_marker(pr) + + assert marker is not None + assert marker["stage"] == "draft" + assert marker["created_at"] == "2026-07-01T00:00:00Z" + + +def test_earliest_marker_of_the_winning_stage_owns_the_grace_period( + workflow: ModuleType, + base_pr: dict[str, Any], +) -> None: + pr = { + **base_pr, + "comments": [ + _authored_comment(_marker("warning", 50, "2026-07-01"), "2026-07-01T00:00:00Z"), + _authored_comment(_marker("warning", 45, "2026-07-20"), "2026-07-20T00:00:00Z"), + ], + } + + marker = workflow._lifecycle_marker(pr) + + assert marker is not None + assert marker["created_at"] == "2026-07-01T00:00:00Z" + + +def test_lifecycle_reaches_closure_at_a_seven_day_cadence( + workflow: ModuleType, + base_pr: dict[str, Any], +) -> None: + """Progression is cadence-independent: weekly runs must still close. + + Simulates warn -> draft -> close on a 7-day cadence, appending the marker + each stage emits, which is what a real weekly schedule would accumulate. + """ + pr: dict[str, Any] = { + **base_pr, + "reviewDecision": "REVIEW_REQUIRED", + "mergeStateStatus": "BLOCKED", + "statusCheckRollup": [{"status": "COMPLETED", "conclusion": "FAILURE"}], + "commits": [{"committedDate": "2026-06-01T00:00:00Z"}], + "comments": [], + "labels": [], + "closingIssuesReferences": [], + "body": "short", + "files": [], + } + dates = [ + "2026-07-01", + "2026-07-08", + "2026-07-15", + "2026-07-22", + "2026-07-29", + "2026-08-05", + "2026-08-12", + "2026-08-19", + ] + state: dict[str, Any] | None = None + actions = [] + for as_of in dates: + result, state = workflow._score_pr(pr, as_of, state) + action = result["recommended_action"] + actions.append(action) + if action in workflow.ACTIONABLE_RECOMMENDATIONS: + pr = { + **pr, + "comments": [ + *pr["comments"], + _authored_comment( + workflow._comment_body(result, as_of), + f"{as_of}T00:00:00Z", + ), + ], + } + if action == "propose_close": + # A closed PR leaves the open-PR snapshot; the ladder ends here. + break + + assert "warn_owner" in actions + assert "propose_draft" in actions + assert "propose_close" in actions + assert actions.index("warn_owner") < actions.index("propose_draft") + assert actions.index("propose_draft") < actions.index("propose_close") + # A stage is never recommended twice: no duplicate owner notifications. + for stage_action in ("warn_owner", "propose_draft", "propose_close"): + assert actions.count(stage_action) == 1 + + +def test_warned_pr_holds_instead_of_re_warning_before_the_deadline( + workflow: ModuleType, + warned_pr: dict[str, Any], +) -> None: + """Inside the warning grace period the ladder holds, it does not repeat.""" + pr = { + **warned_pr, + "comments": [ + _authored_comment(_marker("warning", 44, "2026-07-28"), "2026-07-28T00:00:00Z") + ], + } + state = { + "last_as_of": "2026-07-28", + "last_score": 44, + "last_next_actor": "OWNER", + "below60_owner_streak": 2, + } + + result, _ = workflow._score_pr(pr, "2026-07-31", state) + + assert result["recommended_action"] == "await_owner_deadline" + assert "awaiting_warning_grace_period" in result["action_reasons"] + + +def test_owner_response_after_a_marker_switches_to_monitoring( + workflow: ModuleType, + warned_pr: dict[str, Any], +) -> None: + pr = { + **warned_pr, + "commits": [{"committedDate": "2026-07-30T00:00:00Z"}], + } + + result, _ = workflow._score_pr(pr, "2026-07-31", None) + + assert result["recommended_action"] == "monitor_response" + assert "owner_responded" in result["action_reasons"] + + +# -------------------------------------------------------------------------- +# Comment bodies and markers +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("action", "stage"), + [ + ("warn_owner", "warning"), + ("propose_draft", "draft"), + ("second_owner_notification", "draft"), + ("propose_close", "closed"), + ], +) +def test_comment_body_embeds_the_stage_marker( + workflow: ModuleType, + warned_pr: dict[str, Any], + action: str, + stage: str, +) -> None: + item, _ = workflow._score_pr(warned_pr, "2026-07-31", None) + item = {**item, "recommended_action": action} + + body = workflow._comment_body(item, "2026-07-31") + + assert body.startswith("@owner ") + assert _marker(stage, item["score"], "2026-07-31") in body + # Every emitted marker must be parseable by the reader that consumes it. + parsed = workflow.MARKER_RE.search(body) + assert parsed is not None + assert parsed.group(1) == stage + + +def test_escalation_comment_names_the_protection_and_is_parseable( + workflow: ModuleType, + base_pr: dict[str, Any], +) -> None: + protected_pr = { + **base_pr, + "isDraft": True, + "reviewDecision": "REVIEW_REQUIRED", + "commits": [{"committedDate": "2026-06-01T00:00:00Z"}], + "comments": [_authored_comment(_marker("draft", 50, "2026-07-17"), "2026-07-17T00:00:00Z")], + "title": "fix(security): prevent command injection", + "labels": [{"name": "security"}], + } + protected, _ = workflow._score_pr(protected_pr, "2026-07-31", None) + + body = workflow._comment_body(protected, "2026-07-31") + + assert body.startswith("@owner ") + assert "priority P0" in body + assert "" in body + assert workflow.ESCALATION_MARKER_RE.search(body) is not None + + +def test_action_marker_rejects_a_non_enforcement_action(workflow: ModuleType) -> None: + with pytest.raises(ValueError): + workflow._action_marker("observe_again", 40, "2026-07-31") + + +# -------------------------------------------------------------------------- +# Cross-run idempotency: dedupe on stage, never on score/as_of +# -------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("action", "stage"), + [ + ("warn_owner", "warning"), + ("propose_draft", "draft"), + ("second_owner_notification", "draft"), + ("propose_close", "closed"), + ], +) +def test_marker_from_a_prior_run_is_recognized_despite_different_score_and_date( + workflow: ModuleType, + action: str, + stage: str, +) -> None: + """The defect this guards: exact-text dedupe re-posts the same notice. + + The prior marker carries a different score and as_of than this run would + emit, so an exact-string comparison would miss it. + """ + pr = {"comments": [_authored_comment(_marker(stage, 33, "2026-01-01"), "2026-01-01T00:00:00Z")]} + + assert workflow._has_marker_for_action(pr, action) is True + + +def test_escalation_marker_from_a_prior_run_is_recognized(workflow: ModuleType) -> None: + pr = { + "comments": [ + _authored_comment( + "", + "2026-01-01T00:00:00Z", + ) + ] + } + + assert workflow._has_marker_for_action(pr, "escalate_protected_pr") is True + + +def test_marker_for_a_different_stage_does_not_dedupe(workflow: ModuleType) -> None: + pr = { + "comments": [ + _authored_comment(_marker("warning", 44, "2026-07-24"), "2026-07-24T00:00:00Z") + ] + } + + assert workflow._has_marker_for_action(pr, "warn_owner") is True + assert workflow._has_marker_for_action(pr, "propose_draft") is False + assert workflow._has_marker_for_action(pr, "escalate_protected_pr") is False + + +def test_marker_authored_by_someone_else_does_not_dedupe(workflow: ModuleType) -> None: + pr = { + "comments": [ + { + "author": {"login": "owner"}, + "createdAt": "2026-07-24T00:00:00Z", + "body": _marker("warning", 44, "2026-07-24"), + "viewerDidAuthor": False, + } + ] + } + + assert workflow._has_marker_for_action(pr, "warn_owner") is False + + +def test_has_marker_for_action_rejects_unknown_actions(workflow: ModuleType) -> None: + with pytest.raises(ValueError): + workflow._has_marker_for_action({"comments": []}, "observe_again") + + +# -------------------------------------------------------------------------- +# Enforcement branches: every path that can mutate GitHub +# -------------------------------------------------------------------------- + + +def _plan(number: int, action: str, score: int = 40) -> dict[str, Any]: + return {"number": number, "score": score, "recommended_action": action} + + +@pytest.fixture +def enforcement(workflow: ModuleType, monkeypatch: pytest.MonkeyPatch, tmp_path: Path): + """Run ``_apply_recommendations`` against a stubbed gh surface.""" + commands: list[list[str]] = [] + + def _run( + plans: list[dict[str, Any]], + live_pr: dict[str, Any] | None = None, + close_allowlist: set[int] | None = None, + state: dict[str, Any] | None = None, + ) -> list[dict[str, Any]]: + monkeypatch.setattr( + workflow, + "_fetch_pr", + lambda _repo, _number: dict(live_pr or {"state": "OPEN", "comments": []}), + ) + monkeypatch.setattr(workflow, "_run_gh_command", lambda args: commands.append(args)) + return workflow._apply_recommendations( + "owner/repo", + "2026-07-31", + plans, + state or {"prs": {}}, + close_allowlist or set(), + tmp_path / "enforcement.json", + ) + + return _run, commands + + +def test_enforcement_skips_non_open_pr_before_mutation(enforcement) -> None: + run, commands = enforcement + + results = run([_plan(7, "warn_owner")], live_pr={"state": "CLOSED", "comments": []}) + + assert results == [{"number": 7, "planned_action": "warn_owner", "status": "skipped_not_open"}] + assert commands == [] + + +def test_enforcement_is_idempotent_across_runs(enforcement) -> None: + """A prior run's marker (different score/date) must suppress the comment.""" + run, commands = enforcement + live = { + "state": "OPEN", + "comments": [ + _authored_comment(_marker("warning", 12, "2026-01-01"), "2026-01-01T00:00:00Z") + ], + } + + results = run([_plan(7, "warn_owner")], live_pr=live) + + assert results[0]["status"] == "already_applied" + assert commands == [] + + +def test_enforcement_skips_on_live_drift( + workflow: ModuleType, + enforcement, + warned_pr: dict[str, Any], +) -> None: + run, commands = enforcement + live = {**warned_pr, "state": "OPEN"} + + # The plan claims a score the live PR does not reproduce. + results = run([_plan(1, "propose_draft", score=99)], live_pr=live) + + assert results[0]["status"] == "skipped_live_drift" + assert results[0]["live_score"] == 44 + assert commands == [] + + +def test_enforcement_refuses_closure_without_the_allowlist( + workflow: ModuleType, + enforcement, + base_pr: dict[str, Any], +) -> None: + run, commands = enforcement + abandoned_pr = { + **base_pr, + "state": "OPEN", + "isDraft": True, + "reviewDecision": "REVIEW_REQUIRED", + "commits": [{"committedDate": "2026-06-01T00:00:00Z"}], + "comments": [_authored_comment(_marker("draft", 50, "2026-07-17"), "2026-07-17T00:00:00Z")], + } + planned, _ = workflow._score_pr(abandoned_pr, "2026-07-31", None) + assert planned["recommended_action"] == "propose_close" + + results = run([planned], live_pr=abandoned_pr, close_allowlist=set()) + + assert results[0]["status"] == "skipped_closure_not_allowlisted" + assert commands == [] + + +def test_enforcement_closes_when_allowlisted( + workflow: ModuleType, + enforcement, + base_pr: dict[str, Any], +) -> None: + run, commands = enforcement + abandoned_pr = { + **base_pr, + "state": "OPEN", + "isDraft": True, + "reviewDecision": "REVIEW_REQUIRED", + "commits": [{"committedDate": "2026-06-01T00:00:00Z"}], + "comments": [_authored_comment(_marker("draft", 50, "2026-07-17"), "2026-07-17T00:00:00Z")], + } + planned, _ = workflow._score_pr(abandoned_pr, "2026-07-31", None) + + results = run([planned], live_pr=abandoned_pr, close_allowlist={1}) + + assert results[0]["status"] == "closed_and_commented" + assert len(commands) == 1 + assert commands[0][:5] == ["pr", "close", "1", "--repo", "owner/repo"] + assert _marker("closed", 25, "2026-07-31") in commands[0][-1] + + +def test_enforcement_drafts_then_comments( + workflow: ModuleType, + enforcement, + warned_pr: dict[str, Any], +) -> None: + run, commands = enforcement + live = {**warned_pr, "state": "OPEN"} + planned, _ = workflow._score_pr(live, "2026-07-31", None) + assert planned["recommended_action"] == "propose_draft" + + results = run([planned], live_pr=live) + + assert results[0]["status"] == "drafted_and_commented" + assert commands[0] == ["pr", "ready", "1", "--repo", "owner/repo", "--undo"] + assert commands[1][:5] == ["pr", "comment", "1", "--repo", "owner/repo"] + assert _marker("draft", 44, "2026-07-31") in commands[1][-1] + + +def test_already_draft_pr_gets_a_second_notification_not_a_draft_call( + workflow: ModuleType, + enforcement, + warned_pr: dict[str, Any], +) -> None: + """A draft PR takes the second-notification path; it is never re-drafted.""" + run, commands = enforcement + live = {**warned_pr, "state": "OPEN", "isDraft": True} + planned, _ = workflow._score_pr(live, "2026-07-31", None) + assert planned["recommended_action"] == "second_owner_notification" + + results = run([planned], live_pr=live) + + assert results[0]["status"] == "commented" + assert [command[:2] for command in commands] == [["pr", "comment"]] + assert _marker("draft", planned["score"], "2026-07-31") in commands[0][-1] + + +def test_enforcement_comments_for_a_warning( + workflow: ModuleType, + enforcement, + at_risk_pr: dict[str, Any], +) -> None: + run, commands = enforcement + live = {**at_risk_pr, "state": "OPEN"} + state = { + "prs": { + "1": { + "last_as_of": "2026-07-30", + "last_score": 56, + "last_next_actor": "OWNER", + "below60_owner_streak": 1, + } + } + } + planned, _ = workflow._score_pr(live, "2026-07-31", state["prs"]["1"]) + assert planned["recommended_action"] == "warn_owner" + + results = run([planned], live_pr=live, state=state) + + assert results[0]["status"] == "commented" + assert commands[0][:5] == ["pr", "comment", "1", "--repo", "owner/repo"] + assert _marker("warning", 56, "2026-07-31") in commands[0][-1] + + +def test_enforcement_records_gh_failures_without_aborting_the_run( + workflow: ModuleType, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, + at_risk_pr: dict[str, Any], +) -> None: + live = {**at_risk_pr, "state": "OPEN"} + previous = { + "last_as_of": "2026-07-30", + "last_score": 56, + "last_next_actor": "OWNER", + "below60_owner_streak": 1, + } + planned, _ = workflow._score_pr(live, "2026-07-31", previous) + + monkeypatch.setattr(workflow, "_fetch_pr", lambda _repo, number: {**live, "number": number}) + + def _boom(_args: list[str]) -> str: + raise RuntimeError("gh command failed (1): rate limited") + + monkeypatch.setattr(workflow, "_run_gh_command", _boom) + + results = workflow._apply_recommendations( + "owner/repo", + "2026-07-31", + [planned, {**planned, "number": 2}], + {"prs": {"1": previous, "2": previous}}, + set(), + tmp_path / "enforcement.json", + ) + + assert [result["status"] for result in results] == ["error", "error"] + assert "rate limited" in results[0]["error"] + + +def test_enforcement_ignores_non_actionable_recommendations(enforcement) -> None: + run, commands = enforcement + + results = run([_plan(7, "observe_again"), _plan(8, "monitor_response")]) + + assert results == [] + assert commands == [] + + +def test_enforcement_journal_is_written_incrementally( + workflow: ModuleType, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + journal = tmp_path / "enforcement.json" + monkeypatch.setattr( + workflow, + "_fetch_pr", + lambda _repo, _number: {"state": "CLOSED", "comments": []}, + ) + + workflow._apply_recommendations( + "owner/repo", + "2026-07-31", + [_plan(7, "warn_owner"), _plan(9, "warn_owner")], + {"prs": {}}, + set(), + journal, + ) + + written = workflow._read_object(journal) + assert [entry["number"] for entry in written["results"]] == [7, 9] + + +# -------------------------------------------------------------------------- +# Input validation +# -------------------------------------------------------------------------- + + +def test_close_allowlist_parsing(workflow: ModuleType) -> None: + assert workflow._parse_close_allowlist("") == set() + assert workflow._parse_close_allowlist("222, 231,222") == {222, 231} + + +@pytest.mark.parametrize("value", ["0", "-1", "abc", "12a", "1,,2", "1.5"]) +def test_close_allowlist_rejects_non_pr_numbers(workflow: ModuleType, value: str) -> None: + with pytest.raises(ValueError): + workflow._parse_close_allowlist(value) + + +@pytest.mark.parametrize("snapshot_id", [".", ".."]) +def test_reserved_snapshot_ids_are_rejected( + workflow: ModuleType, + monkeypatch: pytest.MonkeyPatch, + snapshot_id: str, +) -> None: + """``..`` matches SAFE_ID_RE but would resolve artifact_dir to the state root.""" + assert workflow.SAFE_ID_RE.fullmatch(snapshot_id) is not None + assert snapshot_id in workflow.RESERVED_IDS + + def _no_network(*_args: Any, **_kwargs: Any) -> Any: + raise AssertionError("validation must reject the id before any gh call") + + monkeypatch.setattr(workflow, "_run_gh", _no_network) + monkeypatch.setattr(workflow, "_run_gh_command", _no_network) + + with pytest.raises(ValueError, match="snapshot_id"): + workflow._run_locked( + { + "repo": "owner/repo", + "as_of": "2026-07-31", + "snapshot_id": snapshot_id, + "mode": "dry_run", + "importance_provider": "claude_code", + "importance_agent": "reviewer", + } + ) + + +# -------------------------------------------------------------------------- +# Per-repository lock +# -------------------------------------------------------------------------- + + +def test_repo_lock_is_exclusive_and_released( + workflow: ModuleType, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + import fcntl + + monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path)) + + handle, lock_path = workflow._acquire_repo_lock("owner/repo") + assert lock_path.is_file() + + contender = lock_path.open("a+", encoding="utf-8") + try: + with pytest.raises(BlockingIOError): + fcntl.flock(contender.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + + fcntl.flock(handle.fileno(), fcntl.LOCK_UN) + handle.close() + + # Released: the contender can now take it. + fcntl.flock(contender.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(contender.fileno(), fcntl.LOCK_UN) + finally: + contender.close() + + +def test_main_releases_the_lock_even_when_the_run_fails( + workflow: ModuleType, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + import fcntl + + monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path)) + monkeypatch.setattr(workflow, "get_inputs", lambda: {"repo": "owner/repo"}) + + def _boom(_inputs: dict[str, Any]) -> None: + raise RuntimeError("run failed") + + monkeypatch.setattr(workflow, "_run_locked", _boom) + + with pytest.raises(RuntimeError, match="run failed"): + workflow.main() + + lock_path = ( + tmp_path + / ".local" + / "state" + / "cao" + / "pr-health" + / workflow._repo_storage_key("owner/repo") + / ".workflow.lock" + ) + contender = lock_path.open("a+", encoding="utf-8") + try: + fcntl.flock(contender.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) + fcntl.flock(contender.fileno(), fcntl.LOCK_UN) + finally: + contender.close() + + +# -------------------------------------------------------------------------- +# Scheduled flows +# -------------------------------------------------------------------------- + + +def test_guard_enforces_exact_fourteen_day_cadence(guard: ModuleType) -> None: + assert guard.is_due(date(2026, 1, 5)) + assert not guard.is_due(date(2026, 1, 12)) + assert guard.is_due(date(2026, 1, 19)) + assert guard.is_due(date(2027, 1, 18)) + assert not guard.is_due(date(2025, 12, 22)) # before the anchor + + +def test_guard_uses_utc_so_thresholds_do_not_shift_at_midnight( + guard: ModuleType, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """``as_of`` is compared against gh's UTC timestamps, so it must be UTC. + + ``date.today()`` is stubbed to a sentinel the UTC clock can never return, + so this fails on a UTC-local machine too — where simply comparing against + ``datetime.now(timezone.utc).date()`` would pass vacuously. + """ + + class _LocalDate(date): + @classmethod + def today(cls) -> date: + return date(1999, 12, 31) + + monkeypatch.setattr(guard, "date", _LocalDate) + + assert guard.today_utc() == datetime.now(timezone.utc).date() + assert guard.today_utc() != date(1999, 12, 31) + + +def test_scheduled_flow_defaults_to_non_mutating_mode() -> None: + flow_path = EXAMPLE_DIR / "pr-health-biweekly.md" + metadata, prompt = _parse_flow_file(flow_path) + + assert metadata["schedule"] == "0 9 * * 0" + assert metadata["script"] == "./pr_health_biweekly_guard.py" + assert (flow_path.parent / metadata["script"]).is_file() + assert "--input mode=dry_run" in prompt + assert "--input close_allowlist=" in prompt + assert "--input mode=apply" not in prompt + + +def test_apply_schedule_requires_explicit_template_and_disables_closure() -> None: + flow_path = EXAMPLE_DIR / "pr-health-biweekly-apply.md" + metadata, prompt = _parse_flow_file(flow_path) + + assert metadata["schedule"] == "0 9 * * 0" + assert metadata["script"] == "./pr_health_biweekly_guard.py" + assert "--input mode=apply" in prompt + assert "--input close_allowlist=" in prompt + assert "Closure is not authorized" in prompt + assert "STANDING UNATTENDED WRITE GRANT" in prompt + + +def _guard_payload() -> dict[str, Any]: + """The guard's real stdout, parsed the way flow_service parses it.""" + import json + import subprocess + import sys + + completed = subprocess.run( + [sys.executable, str(EXAMPLE_DIR / "pr_health_biweekly_guard.py")], + capture_output=True, + text=True, + check=True, + ) + payload: dict[str, Any] = json.loads(completed.stdout) + assert set(payload) == {"execute", "output"} + return payload + + +def _rendered_flows() -> dict[str, str]: + """Both flow prompts rendered from the guard's actual output. + + Rendering from the guard's real payload — rather than a hand-written + variable dict — is what makes these assertions able to fail if the guard + stops differentiating identifiers by mode. + """ + from cli_agent_orchestrator.utils.template import render_template + + variables = _guard_payload()["output"] + return { + name: render_template(_parse_flow_file(EXAMPLE_DIR / name)[1], variables) + for name in ("pr-health-biweekly.md", "pr-health-biweekly-apply.md") + } + + +def test_guard_emits_every_variable_both_flow_templates_require() -> None: + """render_template raises on a missing variable, so this is the wiring proof.""" + assert set(_rendered_flows()) == { + "pr-health-biweekly.md", + "pr-health-biweekly-apply.md", + } + + +def test_both_flows_can_be_registered_without_colliding() -> None: + """Mode-agnostic identifiers would make the two flows mutually exclusive. + + The second flow to run on a due Monday would hit the workflow's manifest + guard ("snapshot_id already exists with different inputs"). + """ + rendered = _rendered_flows() + + def _value(text: str, flag: str) -> str: + return text.split(f"--input {flag}=", 1)[1].split()[0] + + def _run_id(text: str) -> str: + return text.split("--run-id ", 1)[1].split()[0] + + dry = rendered["pr-health-biweekly.md"] + apply_ = rendered["pr-health-biweekly-apply.md"] + + assert _value(dry, "snapshot_id") != _value(apply_, "snapshot_id") + assert _run_id(dry) != _run_id(apply_) + # Both must still agree on the repository and the evaluation date. + assert _value(dry, "repo") == _value(apply_, "repo") + assert _value(dry, "as_of") == _value(apply_, "as_of")