DEVX-155: Replace pr_review with spec-driven CI gates and pr-review skill

src/devx/ci/check_pr_size.py

@@ -0,0 +1,211 @@
+#!/usr/bin/env python3
+# Implements: REQ-2
+"""Check PR size and reject oversized PRs.
+
+Enforces max lines changed and max files changed to keep PRs small
+and deployable. Generated/excluded files are not counted.
+
+PRs with the ``refactoring`` label bypass the size check — large but
+legitimate refactoring PRs that touch many files in a coordinated way.
+
+Usage:
+    python -m devx.ci.check_pr_size --base origin/master --head HEAD
+
+In CI, pass ``--github-output`` to set ``pr-size-ok`` and ``pr-size-detail``
+for downstream steps.
+"""
+
+from __future__ import annotations
+
+import subprocess  # nosec B404
+
+import click
+from dotenv import load_dotenv
+
+from devx.api_clients import GiteaClient
+from devx.ci._shared import write_github_output
+from devx.config import GITEA_API_URL
+from devx.i18n import _
+from devx.tokens import get_ci_token
+
+load_dotenv()
+
+# Files/patterns excluded from size counting (generated, badges, locks, etc.)
+DEFAULT_EXCLUDED_PATTERNS = [
+    "CHANGELOG.md",
+    "README.md",
+    "docs/index.md",
+    "*.svg",
+    "uv.lock",
+    "poetry.lock",
+    "Pipfile.lock",
+    "package-lock.json",
+    "yarn.lock",
+    "go.sum",
+]
+
+DEFAULT_MAX_LINES = 500
+DEFAULT_MAX_FILES = 10
+REFACTORING_LABEL = "refactoring"
+
+
+def has_refactoring_label(repo: str, pr_number: int) -> bool:
+    """Check if a PR has the 'refactoring' label (bypasses size check)."""
+    try:
+        token = get_ci_token()
+        owner, repo_name = repo.split("/", 1)
+        client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
+        pr = client.get_pr(pr_number)
+        labels = pr.get("labels", [])
+        return any(label.get("name") == REFACTORING_LABEL for label in labels)
+    except Exception:
+        return False
+
+
+def get_diff_stats(base: str, head: str) -> list[tuple[str, int, int]]:
+    """Get per-file diff stats (additions, deletions) between base and head.
+
+    Returns a list of (filename, additions, deletions) tuples.
+    """
+    result = subprocess.run(  # nosec B603 B607
+        ["git", "diff", "--numstat", base, head],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        raise click.ClickException(_("git diff --numstat failed: {stderr}", stderr=result.stderr.strip()))
+    stats: list[tuple[str, int, int]] = []
+    for line in result.stdout.strip().split("\n"):
+        if not line:
+            continue
+        parts = line.split("\t")
+        if len(parts) != 3:
+            continue
+        additions_s, deletions_s, filename = parts
+        # Binary files show "-" for additions/deletions
+        additions = int(additions_s) if additions_s.isdigit() else 0
+        deletions = int(deletions_s) if deletions_s.isdigit() else 0
+        stats.append((filename, additions, deletions))
+    return stats
+
+
+def is_excluded(filename: str, excluded_patterns: list[str]) -> bool:
+    """Check if a filename matches any excluded pattern."""
+    from fnmatch import fnmatch
+
+    return any(fnmatch(filename, pat) for pat in excluded_patterns)
+
+
+def check_size(
+    stats: list[tuple[str, int, int]],
+    max_lines: int,
+    max_files: int,
+    excluded_patterns: list[str],
+) -> tuple[bool, str]:
+    """Check diff stats against limits.
+
+    Returns (is_ok, detail_message).
+    """
+    included = [(f, a, d) for f, a, d in stats if not is_excluded(f, excluded_patterns)]
+    total_lines = sum(a + d for _, a, d in included)
+    total_files = len(included)
+
+    if total_files == 0:
+        return True, "No non-excluded files changed"
+
+    if total_files > max_files:
+        return False, _(
+            "PR has {file_count} files changed (max {max_files}). Excluded: {excluded_count} files.",
+            file_count=total_files,
+            max_files=max_files,
+            excluded_count=len(stats) - total_files,
+        )
+
+    if total_lines > max_lines:
+        return False, _(
+            "PR has {line_count} lines changed (max {max_lines}). Excluded: {excluded_count} files.",
+            line_count=total_lines,
+            max_lines=max_lines,
+            excluded_count=len(stats) - total_files,
+        )
+
+    return True, _(
+        "PR size OK: {file_count} files, {line_count} lines (max {max_files} files, {max_lines} lines).",
+        file_count=total_files,
+        line_count=total_lines,
+        max_files=max_files,
+        max_lines=max_lines,
+    )
+
+
+@click.command()
+@click.option("--base", default="origin/master", help=_("Base ref for diff"))
+@click.option("--head", default="HEAD", help=_("Head ref for diff"))
+@click.option(
+    "--max-lines",
+    type=int,
+    default=DEFAULT_MAX_LINES,
+    help=_("Max lines changed (excluded files not counted)"),
+)
+@click.option(
+    "--max-files",
+    type=int,
+    default=DEFAULT_MAX_FILES,
+    help=_("Max files changed (excluded files not counted)"),
+)
+@click.option(
+    "--github-output",
+    "github_output",
+    is_flag=True,
+    default=False,
+    help=_("Write results to $GITHUB_OUTPUT"),
+)
+@click.option(
+    "--excluded",
+    "excluded",
+    multiple=True,
+    help=_("Additional excluded patterns (in addition to defaults)"),
+)
+@click.option("--repo", default=None, help=_("Repo (owner/name) for label check"))
+@click.option("--pr-number", type=int, default=None, help=_("PR number for label check"))
+def cli(
+    base: str,
+    head: str,
+    max_lines: int,
+    max_files: int,
+    github_output: bool,
+    excluded: tuple[str, ...],
+    repo: str | None,
+    pr_number: int | None,
+) -> None:
+    """Check PR size and reject oversized PRs."""
+    # Check for refactoring label bypass
+    if repo and pr_number and has_refactoring_label(repo, pr_number):
+        detail = _("PR has 'refactoring' label — size check bypassed.")
+        if github_output:
+            write_github_output("pr-size-ok", "true")
+            write_github_output("pr-size-detail", detail)
+        click.echo(f"[pr-size] {detail}")
+        return
+
+    excluded_patterns = list(DEFAULT_EXCLUDED_PATTERNS) + list(excluded)
+    stats = get_diff_stats(base, head)
+    is_ok, detail = check_size(stats, max_lines, max_files, excluded_patterns)
+
+    if github_output:
+        write_github_output("pr-size-ok", "true" if is_ok else "false")
+        write_github_output("pr-size-detail", detail)
+
+    if is_ok:
+        click.echo(f"[pr-size] {detail}")
+    else:
+        click.echo(f"[pr-size] FAILED: {detail}", err=True)
+        click.echo("", err=True)
+        click.echo("Oversized PRs cannot be reliably reviewed or deployed independently.", err=True)
+        click.echo("Split your work into smaller PRs, each addressing one concern.", err=True)
+        raise click.ClickException(_("PR size check failed."))
+
+
+if __name__ == "__main__":  # pragma: no cover
+    cli()

src/devx/ci/create_dependency_pr.py

@@ -0,0 +1,226 @@
+#!/usr/bin/env python3
+# Implements: REQ-5
+"""Auto-create an infra PR to bump a pinned dependency version.
+
+After grm or sso-bridge publishes a new package version, this module
+creates a PR in the infra repo to bump the pinned version in
+``pyproject.toml`` or ``ansible/group_vars/all/images.yml``.
+
+Reuses ``devx.tools.create_pr`` for PR creation and Vikunja task linking.
+
+Usage:
+    python -m devx.ci.create_dependency_pr \
+        --repo oblachno/infra \
+        --package grm \
+        --new-version 0.5.2 \
+        --source-repo oblachno/grm \
+        --source-run-id 12345
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess  # nosec B404
+from pathlib import Path
+
+import click
+from dotenv import load_dotenv
+
+from devx.api_clients import GiteaClient
+from devx.config import GITEA_API_URL, VIKUNJA_API_URL, VIKUNJA_PROJECT_ID
+from devx.exceptions import APIError
+from devx.i18n import _
+from devx.tokens import get_ci_token, get_vikunja_token
+from devx.tools.create_pr import find_existing_pr
+
+load_dotenv()
+
+# Where infra pins dependency versions
+PYPROJECT_PATH = "pyproject.toml"
+IMAGES_YML_PATH = "ansible/group_vars/all/images.yml"
+
+
+def find_pinned_version(package: str, file_path: str) -> str | None:
+    """Find the currently pinned version of a package in a file.
+
+    Looks for patterns like:
+    - ``"grm @ git+...@v0.5.1"``
+    - ``grm = "0.5.1"``
+    - ``grm_version: "0.5.1"``
+    - ``grm_image_version: "0.5.1"``
+    """
+    path = Path(file_path)
+    if not path.exists():
+        return None
+    content = path.read_text(encoding="utf-8")
+    # Match various pinning patterns
+    patterns = [
+        rf"{package}\s*@\s*git\+[^@]+@v?([\d.]+)",  # pip: package @ git+url@vX.Y.Z
+        rf'{package}\s*=\s*"([\d.]+)"',  # pyproject: package = "X.Y.Z"
+        rf'{package}_version:\s*"([\d.]+)"',  # ansible vars: package_version: "X.Y.Z"
+        rf'{package}_image_version:\s*"([\d.]+)"',  # ansible vars: package_image_version: "X.Y.Z"
+    ]
+    for pat in patterns:
+        match = re.search(pat, content)
+        if match:
+            return match.group(1)
+    return None
+
+
+def update_pinned_version(file_path: str, package: str, old_version: str, new_version: str) -> bool:
+    """Update the pinned version in a file. Returns True if changed."""
+    path = Path(file_path)
+    if not path.exists():
+        return False
+    content = path.read_text(encoding="utf-8")
+    # Replace old version with new version in package-related lines
+    patterns = [
+        (rf"({package}\s*@\s*git\+[^@]+@v?){old_version}", rf"\g<1>{new_version}"),
+        (rf'({package}\s*=\s*"){old_version}(")', rf"\g<1>{new_version}\g<2>"),
+        (rf'({package}_version:\s*"){old_version}(")', rf"\g<1>{new_version}\g<2>"),
+        (rf'({package}_image_version:\s*"){old_version}(")', rf"\g<1>{new_version}\g<2>"),
+    ]
+    new_content = content
+    changed = False
+    for pat, replacement in patterns:
+        new_content, n = re.subn(pat, replacement, new_content)
+        if n > 0:
+            changed = True
+    if changed:
+        path.write_text(new_content, encoding="utf-8")
+    return changed
+
+
+def create_vikunja_task(title: str, description: str) -> str | None:
+    """Create a Vikunja task and return its identifier (e.g., OBL-INFRA-531)."""
+    try:
+        token = get_vikunja_token()
+    except click.ClickException:
+        return None
+
+    from devx.api_clients import VikunjaClient
+
+    client = VikunjaClient(VIKUNJA_API_URL, token)
+    task = client.create_task(VIKUNJA_PROJECT_ID, title=title, description=description)
+    return str(task.get("identifier", ""))
+
+
+@click.command()
+@click.option("--repo", default="oblachno/infra", help=_("Target repo (owner/name) to create PR in"))
+@click.option("--package", required=True, help=_("Package name to bump (e.g., grm, sso-bridge)"))
+@click.option("--new-version", required=True, help=_("New version to pin"))
+@click.option("--source-repo", required=True, help=_("Source repo that published (owner/name)"))
+@click.option("--source-run-id", default="", help=_("CI run ID that triggered the publish"))
+@click.option("--dry-run", is_flag=True, default=False, help=_("Show what would be done without creating PR"))
+def cli(
+    repo: str,
+    package: str,
+    new_version: str,
+    source_repo: str,
+    source_run_id: str,
+    dry_run: bool,
+) -> None:
+    """Create an infra PR to bump a pinned dependency version."""
+    token = get_ci_token()
+    if "/" not in repo:
+        raise click.ClickException(_("Invalid repo format: {repo}", repo=repo))
+    owner, repo_name = repo.split("/", 1)
+    client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
+
+    # Find current pinned version
+    old_version = None
+    changed_file = None
+    for f in [PYPROJECT_PATH, IMAGES_YML_PATH]:
+        old_version = find_pinned_version(package, f)
+        if old_version:
+            changed_file = f
+            break
+
+    if not old_version:
+        click.echo(_("[dep-pr] Could not find pinned version for {pkg} in infra repo.", pkg=package))
+        if dry_run:
+            return
+        raise click.ClickException(_("Could not find pinned version for {pkg}", pkg=package))
+
+    if old_version == new_version:
+        click.echo(_("[dep-pr] {pkg} already at {version} — no PR needed.", pkg=package, version=new_version))
+        return
+
+    click.echo(
+        _(
+            "[dep-pr] Bumping {pkg} from {old} to {new} in {file}",
+            pkg=package,
+            old=old_version,
+            new=new_version,
+            file=changed_file,
+        )
+    )
+
+    if dry_run:
+        click.echo(f"[dep-pr] DRY RUN: would update {changed_file} and create PR")
+        return
+
+    # Create a branch
+    branch_name = f"deps/{package}-{new_version}"
+    base_branch = "master"
+
+    # Check for existing PR (reuse from tools.create_pr)
+    existing = find_existing_pr(client, branch_name)
+    if existing:
+        click.echo(_("[dep-pr] PR already exists: #{number}", number=existing.get("number", "?")))
+        return
+
+    # Create branch via API
+    try:
+        master_ref = client._request("GET", "/git/refs/heads/master").json()
+        master_sha = master_ref.get("object", {}).get("sha", "")
+        if not master_sha:
+            raise click.ClickException("Could not get master SHA")
+        client._request("POST", "/git/refs", json={"ref": f"refs/heads/{branch_name}", "sha": master_sha})
+    except APIError as e:
+        if "already exists" in str(e).lower():
+            click.echo(f"[dep-pr] Branch {branch_name} already exists")
+        else:
+            raise click.ClickException(_("Failed to create branch: {error}", error=str(e))) from None
+
+    # Clone, update file, commit, push
+    subprocess.run(["git", "fetch", "origin", f"{branch_name}"], check=False, capture_output=True)  # nosec B603 B607
+    subprocess.run(["git", "checkout", branch_name], check=False, capture_output=True)  # nosec B603 B607
+
+    if not changed_file or not update_pinned_version(changed_file, package, old_version, new_version):
+        raise click.ClickException(_("Failed to update {file}", file=changed_file))
+
+    subprocess.run(["git", "add", changed_file], check=True)  # nosec B603 B607
+    commit_msg = f"deps: bump {package} from {old_version} to {new_version}"
+    subprocess.run(["git", "commit", "-m", commit_msg], check=True)  # nosec B603 B607
+    subprocess.run(["git", "push", "origin", branch_name], check=True)  # nosec B603 B607
+
+    # Create Vikunja task for tracking
+    task_title = f"Bump {package} to {new_version}"
+    task_desc = (
+        f"<p>Auto-created dependency bump PR.</p>"
+        f"<p>Package: {package}</p>"
+        f"<p>Version: {old_version} → {new_version}</p>"
+        f"<p>Source: {source_repo} (run #{source_run_id})</p>"
+    )
+    task_id = create_vikunja_task(task_title, task_desc)
+
+    # Create PR directly (dependency PRs have custom titles, not Vikunja-derived)
+    pr_title = f"{task_id}: {task_title}" if task_id else task_title
+    pr_body = (
+        f"## Dependency Bump\n\n"
+        f"Bumps **{package}** from `{old_version}` to `{new_version}`.\n\n"
+        f"- **Source**: {source_repo}\n"
+        f"- **Triggered by**: CI run #{source_run_id}\n"
+        f"- **Changed file**: `{changed_file}`\n\n"
+        f"This PR was auto-created by `devx.ci.create_dependency_pr`.\n"
+    )
+    if task_id:
+        pr_body += f"\nCloses {task_id}"
+
+    pr = client.create_pr(title=pr_title, head=branch_name, base=base_branch, body=pr_body)
+    click.echo(_("[dep-pr] Created PR #{number}: {title}", number=pr.get("number", "?"), title=pr_title))
+
+
+if __name__ == "__main__":  # pragma: no cover
+    cli()

src/devx/ci/doc_coverage.py

@@ -44,7 +44,6 @@ REQUIRED_SCRIPTS = [
     "auto_merge.py",
     "release.py",
     "publish.py",
-    "pr_review.py",
     "notify_failure.py",
     "post_merge.py",
     "classify_changes.py",

src/devx/ci/fast_molecule.py

@@ -0,0 +1,121 @@
+#!/usr/bin/env python3
+# Implements: REQ-3
+"""Detect changed Ansible roles and output fast molecule test commands.
+
+Reuses ``devx.molecule.molecule_changed`` for role detection (which handles
+playbook→role mapping and shared infrastructure paths).
+
+Fast molecule = converge + verify only, single platform, no idempotence
+check. Used in pre-merge CI to get quick feedback on Ansible changes
+without running the full molecule suite (which runs nightly).
+
+Usage:
+    python -m devx.ci.fast_molecule --base origin/master --head HEAD
+
+Outputs the list of changed roles and the molecule commands to run.
+In CI, pass ``--github-output`` to set ``fast-molecule-roles`` (space-
+separated) and ``fast-molecule-needed`` (true/false) for downstream steps.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import click
+from dotenv import load_dotenv
+
+from devx.ci._shared import write_github_output
+from devx.i18n import _
+from devx.molecule.molecule_changed import detect_changed_roles, get_changed_files
+
+load_dotenv()
+
+
+def get_molecule_scenarios(role_name: str, roles_dir: str = "ansible/roles") -> list[str]:
+    """Get list of molecule scenario names for a role."""
+    mol_dir = Path(roles_dir) / role_name / "molecule"
+    if not mol_dir.is_dir():
+        return []
+    scenarios = []
+    for p in mol_dir.iterdir():
+        if p.is_dir() and (p / "molecule.yml").exists():
+            scenarios.append(p.name)
+    return sorted(scenarios)
+
+
+def build_molecule_commands(
+    roles: set[str],
+    roles_dir: str = "ansible/roles",
+    platform: str = "ubuntu-2604",
+) -> list[str]:
+    """Build molecule test commands for changed roles.
+
+    For each role, runs each scenario with converge + verify only
+    (skip create/destroy between scenarios, skip idempotence).
+    """
+    commands: list[str] = []
+    for role in sorted(roles):
+        scenarios = get_molecule_scenarios(role, roles_dir)
+        if not scenarios:
+            continue
+        for scenario in scenarios:
+            cmd = f"molecule test -s {scenario} --destroy=never --platform-name={platform}"
+            commands.append(cmd)
+    return commands
+
+
+@click.command()
+@click.option("--base", default="origin/master", help=_("Base ref for diff"))
+@click.option("--head", default="HEAD", help=_("Head ref for diff"))
+@click.option("--roles-dir", default="ansible/roles", help=_("Directory containing Ansible roles"))
+@click.option("--platform", default="ubuntu-2604", help=_("Single platform to test against"))
+@click.option(
+    "--github-output",
+    "github_output",
+    is_flag=True,
+    default=False,
+    help=_("Write results to $GITHUB_OUTPUT"),
+)
+def cli(
+    base: str,
+    head: str,
+    roles_dir: str,
+    platform: str,
+    github_output: bool,
+) -> None:
+    """Detect changed roles and output fast molecule test commands."""
+    # Use molecule_changed for role detection (handles playbooks, shared infra)
+    files = get_changed_files(base)
+    if not files:
+        click.echo("[fast-molecule] No files changed.")
+        if github_output:
+            write_github_output("fast-molecule-needed", "false")
+            write_github_output("fast-molecule-roles", "")
+        return
+
+    roles = detect_changed_roles(files)
+    if not roles:
+        click.echo("[fast-molecule] No Ansible roles changed.")
+        if github_output:
+            write_github_output("fast-molecule-needed", "false")
+            write_github_output("fast-molecule-roles", "")
+        return
+
+    commands = build_molecule_commands(roles, roles_dir, platform)
+
+    if github_output:
+        write_github_output("fast-molecule-needed", "true" if commands else "false")
+        write_github_output("fast-molecule-roles", " ".join(sorted(roles)))
+
+    click.echo(_("[fast-molecule] Changed roles: {roles}", roles=", ".join(sorted(roles))))
+    if not commands:
+        click.echo("[fast-molecule] No molecule scenarios found for changed roles.")
+        return
+
+    click.echo(f"[fast-molecule] {len(commands)} scenario(s) to run:")
+    for cmd in commands:
+        click.echo(f"  {cmd}")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    cli()

src/devx/ci/nightly_gate.py

@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+# Implements: REQ-4
+"""Check if the nightly CI gate has passed; block staging deploys if it failed.
+
+The nightly gate stores its status as a Gitea Actions repository variable
+named ``NIGHTLY_STATUS`` on the infra repo. Values:
+
+- ``passed`` — nightly molecule + staging deploy + integration tests passed.
+- ``failed:<run_id>`` — nightly failed. Staging deploys are blocked until
+  the nightly passes again.
+- (not set) — nightly hasn't run yet. First deploy is allowed (bootstrap).
+
+Usage:
+    python -m devx.ci.nightly_gate --repo oblachno/infra --action check
+    python -m devx.ci.nightly_gate --repo oblachno/infra --action set-passed --run-id 12345
+    python -m devx.ci.nightly_gate --repo oblachno/infra --action set-failed --run-id 12345
+"""
+
+from __future__ import annotations
+
+import click
+from dotenv import load_dotenv
+
+from devx.api_clients import GiteaClient
+from devx.ci._shared import write_github_output
+from devx.config import GITEA_API_URL
+from devx.i18n import _
+from devx.tokens import get_ci_token
+
+load_dotenv()
+
+NIGHTLY_STATUS_VAR = "NIGHTLY_STATUS"
+
+
+def get_nightly_status(client: GiteaClient) -> str:
+    """Get the nightly status variable. Returns empty string if not set."""
+    val = client.get_repo_variable(NIGHTLY_STATUS_VAR)
+    return val or ""
+
+
+def set_nightly_status(client: GiteaClient, status: str) -> None:
+    """Set the nightly status variable."""
+    client.set_repo_variable(NIGHTLY_STATUS_VAR, status)
+
+
+@click.command()
+@click.option("--repo", required=True, help=_("Repository in owner/name format"))
+@click.option(
+    "--action",
+    type=click.Choice(["check", "set-passed", "set-failed"]),
+    required=True,
+    help=_("Action to perform"),
+)
+@click.option("--run-id", default="", help=_("CI run ID (for set-failed/set-passed)"))
+@click.option(
+    "--github-output",
+    "github_output",
+    is_flag=True,
+    default=False,
+    help=_("Write results to $GITHUB_OUTPUT"),
+)
+def cli(repo: str, action: str, run_id: str, github_output: bool) -> None:
+    """Check or set the nightly CI gate status."""
+    token = get_ci_token()
+    if "/" not in repo:
+        raise click.ClickException(_("Invalid repo format: {repo}. Expected owner/name.", repo=repo))
+    owner, repo_name = repo.split("/", 1)
+    client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
+
+    if action == "check":
+        status = get_nightly_status(client)
+        if not status:
+            # Bootstrap: no nightly has run yet, allow deploy
+            click.echo("[nightly-gate] No nightly status set — allowing deploy (bootstrap).")
+            if github_output:
+                write_github_output("nightly-gate-passed", "true")
+                write_github_output("nightly-status", "")
+            return
+
+        if status.startswith("passed"):
+            click.echo("[nightly-gate] Nightly passed. Deploy allowed.")
+            if github_output:
+                write_github_output("nightly-gate-passed", "true")
+                write_github_output("nightly-status", status)
+        elif status.startswith("failed"):
+            run_part = status.split(":", 1)[1] if ":" in status else ""
+            run_link = f" (run #{run_part})" if run_part else ""
+            click.echo(
+                _(
+                    "[nightly-gate] Nightly FAILED{run}. Staging deploys are blocked until nightly passes.",
+                    run=run_link,
+                ),
+                err=True,
+            )
+            if github_output:
+                write_github_output("nightly-gate-passed", "false")
+                write_github_output("nightly-status", status)
+            raise click.ClickException(_("Nightly gate failed — staging deploy blocked."))
+        else:
+            click.echo(f"[nightly-gate] Unknown nightly status: {status} — allowing deploy.")
+            if github_output:
+                write_github_output("nightly-gate-passed", "true")
+                write_github_output("nightly-status", status)
+
+    elif action == "set-passed":
+        set_nightly_status(client, f"passed:{run_id}" if run_id else "passed")
+        click.echo(_("[nightly-gate] Set NIGHTLY_STATUS=passed{run}", run=f":{run_id}" if run_id else ""))
+        if github_output:
+            write_github_output("nightly-status", f"passed:{run_id}" if run_id else "passed")
+
+    elif action == "set-failed":
+        set_nightly_status(client, f"failed:{run_id}" if run_id else "failed")
+        click.echo(_("[nightly-gate] Set NIGHTLY_STATUS=failed{run}", run=f":{run_id}" if run_id else ""))
+        if github_output:
+            write_github_output("nightly-status", f"failed:{run_id}" if run_id else "failed")
+
+
+if __name__ == "__main__":  # pragma: no cover
+    cli()

src/devx/ci/pr_review.py

@@ -1,715 +0,0 @@
-#!/usr/bin/env python3
-"""Automated PR review: check architecture compliance, best practices, and quality.
-
-Fetches the PR diff via the Gitea API, runs a series of automated checks,
-and posts a structured review using GiteaClient.create_review.
-
-Checks performed:
-  1. Architecture compliance — no business logic in CLI, no direct subprocess
-     calls outside executor, no hardcoded config that should be in config.py
-  2. Best practices — no bare except, no print() (use click.echo), no TODO/FIXME
-     left in merged code, no functions > 50 lines
-  3. Security — no secrets in code, no shell=True, no eval/exec
-  4. i18n — no raw English strings in click.echo() without _() wrapper
-  5. Resource management — no open() without with statement, no subprocess without cleanup
-  6. Documentation — new CLI commands documented, new modules in architecture.md
-  7. Test coverage — 100% enforced by pytest-cov (checked in quality job)
-  8. Commit conventions — conventional commit format on branch commits
-
-Usage:
-  CI_GITEA_API_TOKEN=<token> [REVIEWER_GITEA_API_TOKEN=<token>] python3 -m devx.ci.pr_review <pr_number> <owner/repo>
-"""
-
-from __future__ import annotations
-
-import os
-import re
-from dataclasses import dataclass, field
-from typing import Any
-
-import click
-from dotenv import load_dotenv  # pyright: ignore[reportMissingImports,reportUnknownVariableType]
-
-from devx.api_clients import GiteaClient
-from devx.config import GITEA_API_URL
-from devx.exceptions import APIError
-from devx.i18n import _
-from devx.tokens import get_ci_token, get_reviewer_token
-
-load_dotenv()
-
-# Files that are exempt from certain checks
-WORKFLOW_ONLY_SUFFIXES = (".yml", ".yaml", ".md", ".json", ".toml", ".cfg", ".ini", ".txt")
-PYTHON_SUFFIX = ".py"
-
-# Architecture rules
-CLI_FILE = "src/devx/cli.py"
-EXECUTOR_FILE = "src/devx/executor.py"
-CONFIG_FILE = "src/devx/config.py"
-
-# Patterns that indicate business logic in CLI (should be in runner_manager.py)
-BUSINESS_LOGIC_IN_CLI = [
-    (r"subprocess\.(run|call|Popen|check_output|check_call)", "subprocess call in CLI — delegate to executor.py"),
-    (r"\bos\.system\b", "os.system call in CLI — delegate to executor.py"),
-    (r"\bansible-playbook\b", "ansible-playbook reference in CLI — delegate to executor.py"),
-]
-
-# Patterns that indicate bad practices
-BAD_PRACTICES = [
-    (r"\bprint\s*\(", "print() found — use click.echo() for user output"),
-    (r"\beval\s*\(", "eval() found — security risk, avoid dynamic code execution"),
-    (r"\bexec\s*\(", "exec() found — security risk, avoid dynamic code execution"),
-    (r"shell\s*=\s*True", "shell=True found — security risk, use shell=False with list args"),
-    (r"except\s*:", "bare except found — catch specific exceptions"),
-    (r"except\s+Exception\s*:", "broad Exception catch — catch specific exceptions"),
-    (r"#\s*(TODO|FIXME|HACK|XXX)", "TODO/FIXME found — resolve before merging"),
-]
-
-# Patterns for hardcoded config values that should be in config.py
-HARDCODED_CONFIG = [
-    (r"https?://[a-z]+\.[a-z]+\.[a-z]+", "hardcoded URL — move to config.py with env var override"),
-]
-
-
-@dataclass
-class ReviewResult:
-    """Result of automated review checks."""
-
-    issues: list[dict[str, Any]] = field(default_factory=list)
-    summary: list[str] = field(default_factory=list)
-
-    @property
-    def has_issues(self) -> bool:
-        return bool(self.issues)
-
-    def add_issue(self, file_path: str, line: int, message: str, severity: str = "warning") -> None:
-        self.issues.append(
-            {
-                "path": file_path,
-                "body": f"[{severity}] {message}",
-                "new_position": line,
-            }
-        )
-
-    def add_summary(self, text: str) -> None:
-        self.summary.append(text)
-
-
-def is_python_file(path: str) -> bool:
-    """Check if a file is a Python source file."""
-    return path.endswith(PYTHON_SUFFIX) and not path.startswith("tests/")
-
-
-def is_workflow_only(path: str) -> bool:
-    """Check if a file is workflow/config/docs only (not Python source)."""
-    return path.endswith(WORKFLOW_ONLY_SUFFIXES) or path.startswith((".gitea/", "docs/", "ansible/"))
-
-
-def check_architecture_compliance(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check that changes follow the documented architecture."""
-    for f in files:
-        path = f.get("filename", "")
-        if not is_python_file(path):
-            continue
-
-        patch = f.get("patch", "")
-        if not patch:
-            continue
-
-        lines = patch.split("\n")
-        current_line = 0
-        for line in lines:
-            if line.startswith("@@"):
-                match = re.search(r"\+(\d+)", line)
-                if match:
-                    current_line = int(match.group(1)) - 1
-                continue
-            if line.startswith("+") and not line.startswith("+++"):
-                current_line += 1
-                content = line[1:]
-
-                # Check for business logic in CLI
-                if path == CLI_FILE:
-                    for pattern, msg in BUSINESS_LOGIC_IN_CLI:
-                        if re.search(pattern, content):
-                            result.add_issue(path, current_line, msg, "error")
-
-    if not result.issues:
-        result.add_summary("- Architecture compliance: OK")
-
-
-def check_best_practices(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check for common code quality issues."""
-    for f in files:
-        path = f.get("filename", "")
-        if not is_python_file(path):
-            continue
-
-        patch = f.get("patch", "")
-        if not patch:
-            continue
-
-        lines = patch.split("\n")
-        current_line = 0
-        for line in lines:
-            if line.startswith("@@"):
-                match = re.search(r"\+(\d+)", line)
-                if match:
-                    current_line = int(match.group(1)) - 1
-                continue
-            if line.startswith("+") and not line.startswith("+++"):
-                current_line += 1
-                content = line[1:]
-
-                for pattern, msg in BAD_PRACTICES:
-                    if re.search(pattern, content):
-                        result.add_issue(path, current_line, msg, "warning")
-
-    if not any(i["body"].startswith("[warning]") for i in result.issues):
-        result.add_summary("- Best practices: OK")
-
-
-def check_security(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check for security issues in changed files."""
-    for f in files:
-        path = f.get("filename", "")
-        if not is_python_file(path):
-            continue
-
-        patch = f.get("patch", "")
-        if not patch:
-            continue
-
-        lines = patch.split("\n")
-        current_line = 0
-        for line in lines:
-            if line.startswith("@@"):
-                match = re.search(r"\+(\d+)", line)
-                if match:
-                    current_line = int(match.group(1)) - 1
-                continue
-            if line.startswith("+") and not line.startswith("+++"):
-                current_line += 1
-                content = line[1:]
-
-                # Check for hardcoded secrets
-                secret_re = r'(token|password|secret|key)\s*=\s*["\'][^"\']{8,}["\']'  # nosec B105
-                is_secret = re.search(secret_re, content, re.IGNORECASE)
-                is_comment = content.strip().startswith("#")
-                is_example = "your-" in content or "example" in content
-                if is_secret and not is_comment and not is_example:
-                    result.add_issue(
-                        path,
-                        current_line,
-                        "potential hardcoded secret — use environment variable",
-                        "error",
-                    )
-
-    if not any(i["body"].startswith("[error]") and "secret" in i["body"] for i in result.issues):
-        result.add_summary("- Security: OK")
-
-
-def check_i18n(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check that user-facing strings are wrapped in _().
-
-    Detects ``click.echo()`` calls with raw string literals that are not
-    wrapped in ``_()``. Only checks ``src/`` files, not tests or scripts.
-    """
-    # Pattern: click.echo("...") or click.echo(f"...") without _() wrapper
-    raw_echo_re = re.compile(r'click\.echo\s*\(\s*["\']([^"\']+)["\']')
-    raw_fstring_re = re.compile(r'click\.echo\s*\(\s*f["\']')
-    # Also check click.ClickException and raise with string
-    raw_exception_re = re.compile(r'click\.ClickException\s*\(\s*["\']([^"\']+)["\']')
-
-    for f in files:
-        path = f.get("filename", "")
-        if not is_python_file(path) or not path.startswith("src/"):
-            continue
-
-        patch = f.get("patch", "")
-        if not patch:
-            continue
-
-        lines = patch.split("\n")
-        current_line = 0
-        for line in lines:
-            if line.startswith("@@"):
-                match = re.search(r"\+(\d+)", line)
-                if match:
-                    current_line = int(match.group(1)) - 1
-                continue
-            if line.startswith("+") and not line.startswith("+++"):
-                current_line += 1
-                content = line[1:]
-
-                # Skip comments and docstrings
-                stripped = content.strip()
-                if stripped.startswith("#") or stripped.startswith('"""') or stripped.startswith("'''"):
-                    continue
-
-                # Check for raw strings in click.echo without _()
-                for regex, msg in [
-                    (raw_echo_re, "click.echo() with raw string — wrap in _() for i18n"),
-                    (raw_fstring_re, "click.echo() with f-string — wrap in _() for i18n"),
-                    (raw_exception_re, "ClickException with raw string — wrap in _() for i18n"),
-                ]:
-                    if regex.search(content):
-                        result.add_issue(path, current_line, msg, "warning")
-
-    if not any("i18n" in i["body"] for i in result.issues):
-        result.add_summary("- i18n: OK")
-
-
-def check_resource_management(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check for resource leaks: open() without with, subprocess without cleanup.
-
-    Detects:
-    - ``open()`` calls not in a ``with`` statement
-    - ``subprocess.Popen()`` without ``.wait()`` or ``.communicate()``
-    """
-    # Pattern: open("...") not preceded by "with" on the same line
-    open_re = re.compile(r"(?<!with\s)\bopen\s*\(")
-    popen_re = re.compile(r"subprocess\.Popen\s*\(")
-
-    for f in files:
-        path = f.get("filename", "")
-        if not is_python_file(path):
-            continue
-
-        patch = f.get("patch", "")
-        if not patch:
-            continue
-
-        lines = patch.split("\n")
-        current_line = 0
-        for line in lines:
-            if line.startswith("@@"):
-                match = re.search(r"\+(\d+)", line)
-                if match:
-                    current_line = int(match.group(1)) - 1
-                continue
-            if line.startswith("+") and not line.startswith("+++"):
-                current_line += 1
-                content = line[1:]
-
-                # Skip comments
-                if content.strip().startswith("#"):
-                    continue
-
-                # Check for open() without with
-                if open_re.search(content) and "with " not in content:
-                    result.add_issue(
-                        path, current_line, "open() without with statement — potential resource leak", "warning"
-                    )
-
-                # Check for Popen without communicate/wait on same line
-                if popen_re.search(content) and ".communicate" not in content and ".wait" not in content:
-                    result.add_issue(
-                        path,
-                        current_line,
-                        "subprocess.Popen() without immediate .communicate() or .wait() — ensure cleanup",
-                        "warning",
-                    )
-
-    if not any("resource" in i["body"].lower() for i in result.issues):
-        result.add_summary("- Resource management: OK")
-
-
-def check_function_length(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check that no new function is excessively long (> 50 lines)."""
-    for f in files:
-        path = f.get("filename", "")
-        if not is_python_file(path):
-            continue
-
-        patch = f.get("patch", "")
-        if not patch:
-            continue
-
-        # Count consecutive added lines within a function
-        lines = patch.split("\n")
-        current_line = 0
-        func_start = 0
-        func_name = ""
-        added_in_func = 0
-
-        for line in lines:
-            if line.startswith("@@"):
-                if func_name and added_in_func > 50:
-                    result.add_issue(
-                        path,
-                        func_start,
-                        f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)",
-                        "warning",
-                    )
-                match = re.search(r"\+(\d+)", line)
-                if match:
-                    current_line = int(match.group(1)) - 1
-                func_name = ""
-                added_in_func = 0
-                continue
-            if line.startswith("+") and not line.startswith("+++"):
-                current_line += 1
-                content = line[1:]
-                func_match = re.match(r"\s*def\s+(\w+)\s*\(", content)
-                if func_match:
-                    if func_name and added_in_func > 50:
-                        result.add_issue(
-                            path,
-                            func_start,
-                            f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)",
-                            "warning",
-                        )
-                    func_name = func_match.group(1)
-                    func_start = current_line
-                    added_in_func = 0
-                else:
-                    added_in_func += 1
-            elif line.startswith(" ") or line.startswith("-"):
-                pass  # context or removed line
-
-        # Check last function
-        if func_name and added_in_func > 50:
-            result.add_issue(
-                path,
-                func_start,
-                f"function '{func_name}' adds {added_in_func} lines — consider splitting (> 50 lines)",
-                "warning",
-            )
-
-
-def check_documentation(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check that documentation is updated for relevant changes."""
-    has_src_changes = any(
-        is_python_file(f.get("filename", "")) and f.get("filename", "").startswith("src/") for f in files
-    )
-    has_doc_changes = any(
-        f.get("filename", "").startswith("docs/") or f.get("filename", "") in ("README.md", "AGENTS.md", "CHANGELOG.md")
-        for f in files
-    )
-    has_ansible_changes = any(f.get("filename", "").startswith("ansible/") for f in files)
-    has_tofu_changes = any(f.get("filename", "").startswith("tofu/") for f in files)
-    has_workflow_changes = any(f.get("filename", "").startswith(".gitea/") for f in files)
-
-    # Check for TODO/FIXME in changed docs
-    todo_issues: list[str] = []
-    for f in files:
-        filename = f.get("filename", "")
-        if filename.endswith(".md") and filename.startswith(("docs/", "README", "AGENTS")):
-            # Can't check file content from PR API easily, but flag if patch adds TODO
-            patch = f.get("patch", "")
-            if patch and re.search(r"^\+.*\b(TODO|FIXME|HACK|XXX)\b", patch, re.IGNORECASE):
-                todo_issues.append(f"{filename}: new TODO/FIXME added in documentation")
-
-    if has_src_changes and not has_doc_changes:
-        result.add_summary("- Documentation: WARNING — source files changed but no docs updated")
-    elif has_ansible_changes and not has_doc_changes:
-        result.add_summary("- Documentation: WARNING — Ansible role changed but no docs updated")
-    elif has_tofu_changes and not has_doc_changes:
-        result.add_summary("- Documentation: WARNING — OpenTofu changes but no docs updated")
-    elif has_workflow_changes and not has_doc_changes:
-        result.add_summary("- Documentation: INFO — workflow changes (consider updating CI docs if behavior changed)")
-    else:
-        result.add_summary("- Documentation: OK")
-
-    if todo_issues:
-        for issue in todo_issues:
-            result.add_summary(f"- Documentation: WARNING — {issue}")
-
-
-def check_test_coverage(files: list[dict[str, Any]], result: ReviewResult) -> None:
-    """Check that tests are updated for source changes."""
-    has_src_changes = any(
-        is_python_file(f.get("filename", "")) and f.get("filename", "").startswith("src/") for f in files
-    )
-    has_test_changes = any(f.get("filename", "").startswith("tests/") for f in files)
-
-    if has_src_changes and not has_test_changes:
-        result.add_summary("- Tests: WARNING — source files changed but no test files updated")
-    else:
-        result.add_summary("- Tests: OK")
-
-
-def check_commit_conventions(client: GiteaClient, pr_number: str, result: ReviewResult) -> None:
-    """Check that PR commits follow conventional commit format.
-
-    Verifies that at least one commit on the PR branch matches the
-    conventional commit pattern (type: description). Merge commits
-    and revert commits are exempt.
-    """
-    try:
-        commits = client.get_pr_commits(pr_number)
-    except APIError as e:
-        result.add_summary(f"- Commit conventions: ERROR — could not fetch commits: {e.message}")
-        return
-
-    if not commits:
-        result.add_summary("- Commit conventions: OK (no commits to check)")
-        return
-
-    from devx.config import CONVENTIONAL_RE
-
-    has_conventional = False
-    non_conventional: list[str] = []
-
-    for commit in commits:
-        commit_info = commit.get("commit", {})
-        message = str(commit_info.get("message", "") if isinstance(commit_info, dict) else "").split("\n")[0]
-        # Skip merge commits and revert commits
-        if message.startswith(("Merge", "Revert")):
-            continue
-        if CONVENTIONAL_RE.match(message):
-            has_conventional = True
-        else:
-            non_conventional.append(message[:60])
-
-    if has_conventional:
-        result.add_summary("- Commit conventions: OK")
-    elif non_conventional:
-        result.add_summary(
-            f"- Commit conventions: WARNING — no conventional commit found. "
-            f"Non-conventional commits: {', '.join(non_conventional[:3])}"
-        )
-    else:
-        result.add_summary("- Commit conventions: OK (all commits are merges/reverts)")
-
-
-def run_review(client: GiteaClient, pr_number: str) -> ReviewResult:
-    """Run all review checks and return the result."""
-    result = ReviewResult()
-
-    try:
-        files = client.get_pr_files(pr_number)
-    except APIError as e:
-        result.add_summary(f"- ERROR: Could not fetch PR files: {e.message}")
-        return result
-
-    if not files:
-        result.add_summary("- No files changed in this PR")
-        return result
-
-    # Run all checks
-    check_architecture_compliance(files, result)
-    check_best_practices(files, result)
-    check_security(files, result)
-    check_i18n(files, result)
-    check_resource_management(files, result)
-    check_function_length(files, result)
-    check_documentation(files, result)
-    check_test_coverage(files, result)
-    check_commit_conventions(client, pr_number, result)
-
-    return result
-
-
-def build_review_body(result: ReviewResult) -> str:
-    """Build the review body text from the review result."""
-    lines = ["## Automated PR Review", ""]
-
-    for item in result.summary:
-        lines.append(item)
-
-    if result.issues:
-        lines.append("")
-        lines.append(f"**{len(result.issues)} issue(s) found:**")
-        lines.append("")
-        for issue in result.issues:
-            lines.append(f"- `{issue['path']}:{issue['new_position']}` — {issue['body']}")
-    else:
-        lines.append("")
-        lines.append("No issues found by automated checks.")
-
-    lines.append("")
-    lines.append("---")
-    lines.append("**Auto-merge:** If all CI checks pass, this PR will be merged automatically.")
-
-    return "\n".join(lines)
-
-
-def post_review(client: GiteaClient, pr_number: str, result: ReviewResult) -> dict[str, Any]:
-    """Post the review to the PR.
-
-    Uses REQUEST_CHANGES when issues are found, COMMENT otherwise.
-    Never uses APPROVE — the bot shares the PR author's token, so
-    Gitea rejects self-approval. The actual APPROVE must come from
-    the manual review step.
-    """
-    body = build_review_body(result)
-    event = "REQUEST_CHANGES" if result.has_issues else "COMMENT"
-    comments = result.issues if result.has_issues else []
-
-    return client.create_review(pr_number, event=event, body=body, comments=comments)
-
-
-def _post_manual_review(
-    client: GiteaClient,
-    pr_number: str,
-    event: str,
-    body: str | None,
-    checklist_confirmed: bool,
-    checklist_categories: str | None,
-    dry_run: bool,
-    owner: str | None = None,
-    repo_name: str | None = None,
-) -> None:
-    """Post a manual review with validation for APPROVE events.
-
-    When self-approval is rejected (reviewer token belongs to PR author),
-    falls back to the CI token (different user) if available.
-    """
-    if not body or len(body) < 50:
-        raise click.ClickException(_("Review body must be at least 50 characters."))
-
-    if event == "APPROVE":
-        if not checklist_confirmed:
-            raise click.ClickException(
-                _("--checklist-confirmed is required for APPROVE events."),
-            )
-        cats = [c.strip() for c in (checklist_categories or "").split(",") if c.strip()]
-        cat_nums: list[int] = []
-        for c in cats:
-            try:
-                cat_nums.append(int(c))
-            except ValueError:
-                raise click.ClickException(
-                    _("Invalid checklist category: {cat}. Must be numbers.", cat=c),
-                ) from None
-        if len(cat_nums) < 8:
-            raise click.ClickException(
-                _("--checklist-categories must list at least 8 of 13 categories. Got {count}.", count=len(cat_nums)),
-            )
-
-    click.echo(f"Manual review event: {event}")
-    click.echo(f"Body: {body[:80]}...")
-    if checklist_confirmed:
-        click.echo(f"Checklist confirmed: {checklist_categories}")
-
-    if dry_run:
-        click.echo("\n[dry-run] Review not posted.")
-        return
-
-    try:
-        review = client.create_review(pr_number, event=event, body=body)
-    except APIError as e:
-        if "approve" in e.message.lower() or "422" in str(e.status):
-            # Self-approval not allowed (reviewer token belongs to PR author).
-            # Fall back to CI token (different user) if available.
-            ci_token = os.environ.get("CI_GITEA_API_TOKEN", "").strip()
-            if ci_token and owner and repo_name:
-                click.echo(_("Note: Self-approval not allowed with reviewer token. Retrying with CI token."))
-                ci_client = GiteaClient(GITEA_API_URL, ci_token, owner, repo_name)
-                try:
-                    review = ci_client.create_review(pr_number, event=event, body=body)
-                except APIError:
-                    click.echo(_("Note: CI token also cannot approve. Posting COMMENT instead."))
-                    review = client.create_review(pr_number, event="COMMENT", body=body)
-            else:
-                click.echo(_("Note: Self-approval not allowed. Posting COMMENT instead."))
-                review = client.create_review(pr_number, event="COMMENT", body=body)
-        else:
-            raise
-    review_id = review.get("id", "?")
-    click.echo(
-        _(
-            "\nReview #{review_id} posted on PR #{pr_number} with event '{event}'.",
-            review_id=review_id,
-            pr_number=pr_number,
-            event=event,
-        )
-    )
-
-
-@click.command()
-@click.argument("pr_number")
-@click.argument("repo")
-@click.option("--dry-run", is_flag=True, default=False, help="Print review without posting.")
-@click.option(
-    "--event",
-    type=click.Choice(["APPROVE", "REQUEST_CHANGES", "COMMENT"], case_sensitive=False),
-    default=None,
-    help="Post a manual review with the given event (skips automated checks).",
-)
-@click.option("--body", default=None, help="Review body text (required with --event).")
-@click.option(
-    "--checklist-confirmed",
-    is_flag=True,
-    default=False,
-    help="Attest that REVIEW_CHECKLIST.md categories were checked (required for APPROVE).",
-)
-@click.option(
-    "--checklist-categories",
-    default=None,
-    help="Comma-separated checklist category numbers (required for APPROVE, min 8 of 13).",
-)
-def main(
-    pr_number: str,
-    repo: str,
-    dry_run: bool,
-    event: str | None,
-    body: str | None,
-    checklist_confirmed: bool,
-    checklist_categories: str | None,
-) -> None:
-    """Run automated PR review and post results to Gitea.
-
-    Without --event: runs automated checks and posts COMMENT/REQUEST_CHANGES.
-    With --event: posts a manual review (skips automated checks).
-    """
-    try:
-        token = get_reviewer_token() if (event and event.upper() == "APPROVE") else get_ci_token()
-    except click.ClickException:
-        raise click.ClickException(_("ERROR: CI_GITEA_TOKEN is not set.")) from None
-
-    owner, repo_name = repo.split("/")
-    client = GiteaClient(GITEA_API_URL, token, owner, repo_name)
-
-    if event is not None:
-        _post_manual_review(
-            client,
-            pr_number,
-            event.upper(),
-            body,
-            checklist_confirmed,
-            checklist_categories,
-            dry_run,
-            owner=owner,
-            repo_name=repo_name,
-        )
-        return
-
-    result = run_review(client, pr_number)
-
-    body = build_review_body(result)
-    event = "REQUEST_CHANGES" if result.has_issues else "COMMENT"
-
-    click.echo(f"Review event: {event}")
-    click.echo(f"Issues found: {len(result.issues)}")
-    click.echo("")
-    click.echo(body)
-
-    if dry_run:
-        click.echo("\n[dry-run] Review not posted.")
-        return
-
-    try:
-        review = post_review(client, pr_number, result)
-    except APIError as e:
-        if "approve" in e.message.lower() or "422" in str(e.status):
-            click.echo(_("Note: Self-approval not allowed. Posting COMMENT instead."))
-            review = client.create_review(pr_number, event="COMMENT", body=body)
-        else:
-            raise
-    review_id = review.get("id", "?")
-    click.echo(
-        _(
-            "\nReview #{review_id} posted on PR #{pr_number} with event '{event}' ({num_comments} inline comments).",
-            review_id=review_id,
-            pr_number=pr_number,
-            event=event,
-            num_comments=len(result.issues),
-        )
-    )
-
-
-if __name__ == "__main__":  # pragma: no cover
-    main()

src/devx/ci/validate_spec.py

@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+# Implements: REQ-1
+"""Validate that a PR has a spec file with required sections and acceptance criteria.
+
+Spec-driven development gate. Runs in CI before expensive jobs.
+
+Validates:
+1. A spec file exists at ``docs/specs/<TASK-ID>.md`` (TASK-ID extracted from branch).
+2. The spec contains required sections: Problem, Approach, Test Plan, Deploy Plan, Rollback Plan.
+3. The spec contains REQ-ID lines (``REQ-N: <description>``).
+4. The spec contains an Acceptance Criteria checklist with at least one item.
+5. All acceptance criteria checkboxes are checked (``- [x]``).
+
+Usage:
+    python -m devx.ci.validate_spec --branch OBL-INFRA-531-fix-foo
+
+In CI, also pass ``--github-output`` to set ``spec-valid`` and ``spec-path``
+for downstream steps.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import click
+from dotenv import load_dotenv
+
+from devx.ci._shared import extract_task_id, write_github_output
+from devx.i18n import _
+
+load_dotenv()
+
+REQUIRED_SECTIONS = [
+    "## Problem",
+    "## Approach",
+    "## Test Plan",
+    "## Deploy Plan",
+    "## Rollback Plan",
+    "## Acceptance Criteria",
+]
+
+REQ_ID_RE = re.compile(r"^REQ-\d+:\s+.+", re.MULTILINE)
+AC_CHECKED_RE = re.compile(r"^\s*- \[x\]\s+.+", re.MULTILINE)
+AC_UNCHECKED_RE = re.compile(r"^\s*- \[ \]\s+.+", re.MULTILINE)
+
+
+def find_spec_file(task_id: str, specs_dir: str = "docs/specs") -> Path | None:
+    """Find the spec file for the given task ID.
+
+    Looks for ``docs/specs/<TASK-ID>.md`` (case-insensitive filename).
+    Returns the Path if found, None otherwise.
+    """
+    base = Path(specs_dir)
+    if not base.is_dir():
+        return None
+    # Exact match (case-insensitive)
+    for p in base.glob("*.md"):
+        if p.stem.upper() == task_id.upper():
+            return p
+    return None
+
+
+def validate_spec_content(content: str) -> list[str]:
+    """Validate spec content and return a list of error messages.
+
+    Returns an empty list if the spec is valid.
+    """
+    errors: list[str] = []
+
+    # Check required sections
+    for section in REQUIRED_SECTIONS:
+        if section not in content:
+            errors.append(_("Missing required section: {section}", section=section))
+
+    # Check for at least one REQ-ID
+    req_ids = REQ_ID_RE.findall(content)
+    if not req_ids:
+        errors.append(_("No REQ-ID lines found. Each requirement must be labeled (e.g., 'REQ-1: <description>')."))
+
+    # Check acceptance criteria has at least one item
+    checked = AC_CHECKED_RE.findall(content)
+    unchecked = AC_UNCHECKED_RE.findall(content)
+    if not checked and not unchecked:
+        errors.append(_("Acceptance Criteria section has no checklist items. Add at least one '- [ ] item'."))
+    elif unchecked:
+        errors.append(
+            _(
+                "Acceptance Criteria has {count} unchecked item(s). All AC items must be checked (- [x]) before merge.",
+                count=len(unchecked),
+            )
+        )
+
+    return errors
+
+
+@click.command()
+@click.option("--branch", required=True, help=_("Branch name (e.g., OBL-INFRA-531-fix-foo)"))
+@click.option("--specs-dir", default="docs/specs", help=_("Directory containing spec files"))
+@click.option(
+    "--github-output",
+    "github_output",
+    is_flag=True,
+    default=False,
+    help=_("Write results to $GITHUB_OUTPUT"),
+)
+@click.option("--allow-missing", is_flag=True, default=False, help=_("Allow missing spec (warn only, don't fail)"))
+def cli(branch: str, specs_dir: str, github_output: bool, allow_missing: bool) -> None:
+    """Validate that a spec file exists and has required content."""
+    task_id = extract_task_id(branch)
+    if not task_id:
+        msg = _("No task ID found in branch name '{branch}'. Expected format: <PREFIX>-N-description.", branch=branch)
+        if allow_missing:
+            click.echo(f"WARNING: {msg}")
+            if github_output:
+                write_github_output("spec-valid", "false")
+                write_github_output("spec-path", "")
+            return
+        raise click.ClickException(msg)
+
+    spec_path = find_spec_file(task_id, specs_dir)
+    if spec_path is None:
+        msg = _(
+            "No spec file found for task {task_id} in {dir}/. Expected: {dir}/{task_id}.md",
+            task_id=task_id,
+            dir=specs_dir,
+        )
+        if allow_missing:
+            click.echo(f"WARNING: {msg}")
+            if github_output:
+                write_github_output("spec-valid", "false")
+                write_github_output("spec-path", "")
+            return
+        raise click.ClickException(msg)
+
+    content = spec_path.read_text(encoding="utf-8")
+    errors = validate_spec_content(content)
+
+    if github_output:
+        write_github_output("spec-valid", "true" if not errors else "false")
+        write_github_output("spec-path", str(spec_path))
+
+    if errors:
+        click.echo("", err=True)
+        click.echo("=" * 60, err=True)
+        click.echo(f"Spec validation FAILED for {spec_path}:", err=True)
+        click.echo("=" * 60, err=True)
+        for e in errors:
+            click.echo(f"  - {e}", err=True)
+        raise click.ClickException(_("Spec validation failed."))
+
+    click.echo(_("[spec-check] Spec validated: {path}", path=spec_path))
+
+
+if __name__ == "__main__":  # pragma: no cover
+    cli()

src/devx/cli.py

@@ -116,13 +116,6 @@ def ci_post_merge(args: tuple[str, ...]) -> None:
     _run_module("devx.ci.post_merge", list(args))
 
 
-@ci.command("pr-review")
-@click.argument("args", nargs=-1)
-def ci_pr_review(args: tuple[str, ...]) -> None:
-    """Run automated PR review."""
-    _run_module("devx.ci.pr_review", list(args))
-
-
 @ci.command("publish")
 @click.argument("args", nargs=-1)
 def ci_publish(args: tuple[str, ...]) -> None:

src/devx/make/devx.mak

@@ -109,7 +109,7 @@ devx-ensure-venv:
 	fi
 
 .PHONY: devx-create-task devx-create-pr devx-push devx-push-with-pr devx-check-config
-.PHONY: devx-pr-status devx-pr-logs devx-pr-label devx-pr-review devx-rebase devx-pr-rebase
+.PHONY: devx-pr-status devx-pr-logs devx-pr-label devx-rebase devx-pr-rebase
 .PHONY: devx-configure-gitea-pypi devx-install-tools devx-install-checkmake devx-checkmake
 .PHONY: devx-workflow-lint devx-workflow-dryrun devx-workflow-dryrun-safe devx-workflow-check
 .PHONY: devx-notify-failure devx-install-hooks devx-activate-scripts devx-venv devx-ensure-venv
@@ -171,16 +171,6 @@ devx-pr-label:
 		$(if $(PR),--pr $(PR)) \
 		--label $(or $(LABEL),ready-to-merge)
 
-# Usage: make devx-pr-review PR=42 EVENT=APPROVE BODY="..." CHECKLIST=1,2,3,4,5,6,7,8,9,10,11,12,13
-#        make devx-pr-review PR=42 EVENT=REQUEST_CHANGES BODY="..."
-#        make devx-pr-review PR=42  (auto review)
-devx-pr-review:
-	@$(DEVX_PYTHON) -m devx.ci.pr_review \
-		$(PR) $(DEVX_REPO_OWNER)/$(DEVX_REPO_NAME) \
-		$(if $(EVENT),--event $(EVENT)) \
-		$(if $(BODY),--body "$(BODY)") \
-		$(if $(CHECKLIST),--checklist-confirmed --checklist-categories $(CHECKLIST))
-
 # Rebase current branch onto origin/master and force-push
 # Usage: make devx-rebase
 #        make devx-rebase NO_PUSH=1

src/devx/tokens.py

@@ -2,19 +2,16 @@
 
 Centralizes Gitea/Vikunja token discovery with role-based environment
 variable names and backwards compatibility with the legacy
-``CI_GITEA_TOKEN`` / ``REVIEW_GITEA_TOKEN`` naming convention.
+``CI_GITEA_TOKEN`` naming convention.
 
 Roles:
 - ``CI_GITEA_API_TOKEN``: CI workflows (read actions, post status, merge, etc.)
-- ``REVIEWER_GITEA_API_TOKEN``: PR approval reviews (must be a different user
-  from the PR author for Gitea to accept the review as an approval)
 - ``DEVELOPER_GITEA_API_TOKEN``: local development tools (create-task,
   create-pr, setup, etc.)
 
 Fallbacks:
 - New role names are checked first.
-- Legacy names (``CI_GITEA_TOKEN``, ``REVIEW_GITEA_TOKEN``) are accepted for
-  backwards compatibility.
+- Legacy names (``CI_GITEA_TOKEN``) are accepted for backwards compatibility.
 - If no role-specific token is set, the generic CI tokens are tried last.
 """
 
@@ -28,12 +25,6 @@ from devx.i18n import _
 
 # Token environment variable names, in lookup priority order.
 CI_TOKEN_NAMES = ["CI_GITEA_API_TOKEN", "CI_GITEA_TOKEN"]
-REVIEWER_TOKEN_NAMES = [
-    "REVIEWER_GITEA_API_TOKEN",
-    # Legacy name used before role-based tokens.
-    "REVIEW_GITEA_TOKEN",
-    *CI_TOKEN_NAMES,
-]
 DEVELOPER_TOKEN_NAMES = ["DEVELOPER_GITEA_API_TOKEN", *CI_TOKEN_NAMES]
 
 VIKUNJA_TOKEN_NAMES = ["VIKUNJA_TOKEN"]
@@ -61,11 +52,6 @@ def get_ci_token() -> str:
     return get_token(*CI_TOKEN_NAMES)
 
 
-def get_reviewer_token() -> str:
-    """Resolve the reviewer Gitea API token used for PR approvals."""
-    return get_token(*REVIEWER_TOKEN_NAMES)
-
-
 def get_developer_token() -> str:
     """Resolve the developer Gitea API token used for local tooling."""
     return get_token(*DEVELOPER_TOKEN_NAMES)