Gates

gates

Reusable pre- and post-agent gates.

Gates validate data before and after an AI agent runs. Pre-gates filter input so the agent never sees invalid data. Post-gates validate output to catch dangerous changes before they reach the forge.

All functions are stateless and tracker-agnostic -- they operate on dicts and file lists, not Jira-specific types.

The GATE_REGISTRY maps CLI-friendly names to GateSpec instances. Use resolve_gates() to look up gates by name and validate_gate_env() to check required environment variables before running any gate.

GateSpec(name, fn, phase, required_env=list()) dataclass

Metadata for a registered gate function.

resolve_gates(names)

Look up gate specs by CLI name. Raises SystemExit on unknown names.

Source code in src/agentic_ci/gates.py

def resolve_gates(names: list[str]) -> list[GateSpec]:
    """Look up gate specs by CLI name. Raises SystemExit on unknown names."""
    specs = []
    unknown = []
    for name in names:
        spec = GATE_REGISTRY.get(name)
        if spec is None:
            unknown.append(name)
        else:
            specs.append(spec)
    if unknown:
        available = ", ".join(sorted(GATE_REGISTRY))
        sys.exit(f"Error: unknown gate(s): {', '.join(unknown)}\nAvailable: {available}")
    return specs

validate_gate_env(gates)

Check that all required env vars for the given gates are set.

Fails with a clear error listing every missing variable and which gate needs it.

Source code in src/agentic_ci/gates.py

def validate_gate_env(gates: list[GateSpec]) -> None:
    """Check that all required env vars for the given gates are set.

    Fails with a clear error listing every missing variable and
    which gate needs it.
    """
    missing: dict[str, list[str]] = {}
    for gate in gates:
        for var in gate.required_env:
            if not os.environ.get(var):
                missing.setdefault(var, []).append(gate.name)
    if missing:
        lines = [f"  {var} (needed by: {', '.join(names)})" for var, names in missing.items()]
        sys.exit("Error: missing environment variables for gates:\n" + "\n".join(lines))

check_sensitive_files(changed_files, blocklist=None)

Check if any changed files match a sensitive-file blocklist.

A pattern without a / is matched against the file's basename, so *secret* blocks config/secret.yaml but not a secrets/ directory. A pattern with a / is matched against the full repo-relative path (.agentic-ci/*); fnmatch's * crosses /, so it covers nested files too. A path is blocked when it matches as given or after unquote_git_path() (for git's C-quoted output), so unquoting can only add a match. Matching is case-sensitive.

Returns a list of blocked file paths, as given (empty means all clear).

Source code in src/agentic_ci/gates.py

def check_sensitive_files(
    changed_files: list[str],
    blocklist: list[str] | None = None,
) -> list[str]:
    """Check if any changed files match a sensitive-file blocklist.

    A pattern without a ``/`` is matched against the file's basename, so
    ``*secret*`` blocks ``config/secret.yaml`` but not a ``secrets/``
    directory.  A pattern with a ``/`` is matched against the full
    repo-relative path (``.agentic-ci/*``); fnmatch's ``*`` crosses ``/``,
    so it covers nested files too.  A path is blocked when it matches as
    given or after ``unquote_git_path()`` (for git's C-quoted output), so
    unquoting can only add a match.  Matching is case-sensitive.

    Returns a list of blocked file paths, as given (empty means all clear).
    """
    if blocklist is None:
        blocklist = DEFAULT_SENSITIVE_BLOCKLIST
    path_patterns = [p for p in blocklist if "/" in p]
    name_patterns = [p for p in blocklist if "/" not in p]
    blocked = []
    for filepath in changed_files:
        for path in (filepath, unquote_git_path(filepath)):
            name = PurePosixPath(path).name
            if any(fnmatch.fnmatchcase(name, p) for p in name_patterns) or any(
                fnmatch.fnmatchcase(path, p) for p in path_patterns
            ):
                blocked.append(filepath)
                break
    return blocked

check_commit_identity(commit_info, expected_email)

Verify the commit committer matches the expected bot email.

Source code in src/agentic_ci/gates.py

def check_commit_identity(commit_info: dict, expected_email: str) -> bool:
    """Verify the commit committer matches the expected bot email."""
    actual = commit_info.get("email", "")
    return actual.lower() == expected_email.lower()

log_changed_files(changed_files, ticket_key)

Log changed files for observability.

Source code in src/agentic_ci/gates.py

def log_changed_files(changed_files: list[str], ticket_key: str) -> None:
    """Log changed files for observability."""
    if changed_files:
        log.info(
            "[%s] Agent changed %d files: %s",
            ticket_key,
            len(changed_files),
            ", ".join(display_path(f) for f in changed_files),
        )
    else:
        log.info("[%s] No files changed by agent", ticket_key)

gitleaks_scan(repo_dir, compare_ref='origin/HEAD')

Scan new commits for secrets using gitleaks.

Scans the commits in compare_ref..HEAD with gitleaks' built-in rules. Nothing in repo_dir changes what is detected: config and ignore files in the working tree, gitleaks:allow comments and the GITLEAKS_CONFIG variables are all ignored.

Returns a list of error strings (empty means clean). Requires gitleaks to be installed on PATH. Fails closed: returns an error if gitleaks is missing or times out.

Source code in src/agentic_ci/gates.py

def gitleaks_scan(repo_dir: Path, compare_ref: str = "origin/HEAD") -> list[str]:
    """Scan new commits for secrets using gitleaks.

    Scans the commits in ``compare_ref..HEAD`` with gitleaks' built-in
    rules.  Nothing in ``repo_dir`` changes what is detected: config and
    ignore files in the working tree, ``gitleaks:allow`` comments and the
    ``GITLEAKS_CONFIG`` variables are all ignored.

    Returns a list of error strings (empty means clean).
    Requires ``gitleaks`` to be installed on PATH.  Fails closed:
    returns an error if gitleaks is missing or times out.
    """
    if not shutil.which("gitleaks"):
        log.error("gitleaks not found on PATH — failing closed")
        return ["gitleaks is not installed; secret scan cannot run"]

    out, errors = _gitleaks_git(
        [
            "-C",
            str(repo_dir),
            "rev-parse",
            "--path-format=absolute",
            "--git-common-dir",
            f"{compare_ref}^{{commit}}",
            "HEAD^{commit}",
        ],
        "rev-parse",
    )
    if errors:
        return errors
    lines = out.split()
    if len(lines) != 3:
        log.error("git rev-parse returned unexpected output: %r", out.strip())
        return ["gitleaks pre-check failed: could not resolve the scan range"]
    git_dir, base, head = lines
    scan_range = f"{base}..{head}"

    with tempfile.TemporaryDirectory(prefix="agentic-ci-gitleaks-") as tmp:
        scratch = Path(tmp)
        scan_repo = scratch / "scan.git"
        _, errors = _gitleaks_git(["init", "--quiet", "--bare", str(scan_repo)], "init")
        if errors:
            return errors
        (scan_repo / "objects" / "info" / "alternates").write_text(
            str(Path(git_dir) / "objects") + "\n"
        )

        out, errors = _gitleaks_git(
            ["--git-dir", str(scan_repo), "rev-list", "--count", scan_range], "rev-list"
        )
        if errors:
            return errors
        try:
            commit_count = int(out.strip())
        except ValueError:
            log.error("git rev-list returned non-integer: %r", out.strip())
            return ["gitleaks pre-check failed: could not parse commit count"]

        if commit_count == 0:
            log.info("gitleaks scan skipped: no commits in range %s..HEAD", compare_ref)
            return []

        config = scratch / "gitleaks.toml"
        config.write_text(_GITLEAKS_CONFIG)
        ignore_dir = scratch / "ignore"
        ignore_dir.mkdir()
        env = {k: v for k, v in os.environ.items() if k not in _GITLEAKS_ENV_OVERRIDES}
        env["GIT_NO_REPLACE_OBJECTS"] = "1"
        try:
            result = subprocess.run(
                [
                    "gitleaks",
                    "detect",
                    "--source",
                    str(scan_repo),
                    f"--log-opts={scan_range}",
                    "--config",
                    str(config),
                    "--gitleaks-ignore-path",
                    str(ignore_dir),
                    "--ignore-gitleaks-allow",
                    "--verbose",
                ],
                capture_output=True,
                text=True,
                timeout=GITLEAKS_TIMEOUT,
                cwd=scratch,
                env=env,
            )
        except subprocess.TimeoutExpired:
            log.error("gitleaks timed out after %ds", GITLEAKS_TIMEOUT)
            return [f"gitleaks timed out after {GITLEAKS_TIMEOUT}s; secret scan inconclusive"]

    if result.returncode != 0:
        log.error("gitleaks detected secrets in committed changes")
        return [
            "gitleaks detected potential secrets in committed code. "
            "Review the gitleaks output in the CI job log for details."
        ]

    log.info("gitleaks scan passed: no secrets found")
    return []

filter_comments_by_domain(comments, allowed_domain_re)

Keep only comments from authors whose email matches allowed_domain_re.

Comments with a missing (None) or non-string author_email are dropped.

Source code in src/agentic_ci/gates.py

def filter_comments_by_domain(
    comments: list[dict],
    allowed_domain_re: re.Pattern[str],
) -> list[dict]:
    """Keep only comments from authors whose email matches ``allowed_domain_re``.

    Comments with a missing (None) or non-string ``author_email`` are dropped.
    """
    return [c for c in comments if _email_matches(allowed_domain_re, c.get("author_email"))]

filter_bot_comments(comments, sentinel_phrases)

Remove comments containing any of the given bot sentinel phrases.

A missing (None) or non-string body holds no sentinel, so the comment is kept.

Source code in src/agentic_ci/gates.py

def filter_bot_comments(
    comments: list[dict],
    sentinel_phrases: list[str],
) -> list[dict]:
    """Remove comments containing any of the given bot sentinel phrases.

    A missing (None) or non-string ``body`` holds no sentinel, so the
    comment is kept.
    """
    kept = []
    for c in comments:
        body = c.get("body")
        if not isinstance(body, str):
            body = ""
        if not any(s in body for s in sentinel_phrases):
            kept.append(c)
    return kept

check_description_editors(editors, reporter_email, internal_domain_re)

Check if any description editors are outside the trusted domain.

Validates the reporter (original author) and all changelog editors. Returns a list of untrusted email addresses (empty means all clear). Treats empty or missing emails as untrusted.

Source code in src/agentic_ci/gates.py

def check_description_editors(
    editors: list[str],
    reporter_email: str,
    internal_domain_re: re.Pattern[str],
) -> list[str]:
    """Check if any description editors are outside the trusted domain.

    Validates the reporter (original author) and all changelog editors.
    Returns a list of untrusted email addresses (empty means all clear).
    Treats empty or missing emails as untrusted.
    """
    untrusted: list[str] = []
    if not reporter_email:
        untrusted.append("missing-email:reporter")
    elif not internal_domain_re.search(reporter_email):
        untrusted.append(reporter_email)

    for email in editors:
        normalized = email or "missing-email:unknown"
        if normalized.startswith("missing-email:") or not internal_domain_re.search(normalized):
            if normalized not in untrusted:
                untrusted.append(normalized)

    return untrusted

check_external_reporter(ticket, internal_domain_re, *, external_label)

Check if a ticket was filed by an external reporter.

Returns external_label if the reporter is external and the label is not already present, otherwise None.

Source code in src/agentic_ci/gates.py

def check_external_reporter(
    ticket: dict,
    internal_domain_re: re.Pattern[str],
    *,
    external_label: str,
) -> str | None:
    """Check if a ticket was filed by an external reporter.

    Returns ``external_label`` if the reporter is external and the label
    is not already present, otherwise ``None``.
    """
    labels = ticket.get("labels", [])
    if external_label in labels:
        return None
    if not _email_matches(internal_domain_re, ticket.get("reporter_email")):
        return external_label
    return None

check_label_author_email(author_info, domain_pattern)

Verify that the label's author email matches domain_pattern.

author_info is the dict returned by JiraClient.get_label_author(), expected to contain "found" (bool) and "email" (str) keys.

Returns True if the author was found and the email matches. A missing (None) or non-string email never matches.

Source code in src/agentic_ci/gates.py

def check_label_author_email(
    author_info: dict,
    domain_pattern: re.Pattern[str],
) -> bool:
    """Verify that the label's author email matches ``domain_pattern``.

    ``author_info`` is the dict returned by ``JiraClient.get_label_author()``,
    expected to contain ``"found"`` (bool) and ``"email"`` (str) keys.

    Returns True if the author was found and the email matches.  A missing
    (None) or non-string email never matches.
    """
    if not author_info.get("found"):
        return False
    return _email_matches(domain_pattern, author_info.get("email"))