Skip to content
Code Mower gate

Code Mower gate #9009

Sign in to view logs

Code Mower gate

Code Mower gate #9009

Workflow file for this run

.github/workflows/code-mower-gate.yml at 166013b
# Code Mower workflow template: publish a single merge-gate commit status.
name: Code Mower gate
on:
pull_request_target:
types: [opened, reopened, synchronize, labeled, unlabeled, ready_for_review]
issue_comment:
types: [created, edited, deleted]
workflow_run:
workflows: ["Code Mower Local CLI Audits"]
types: [completed]
workflow_dispatch:
inputs:
pr_number:
description: "Pull request number"
required: true
head_sha:
description: "Pull request head SHA"
required: true
permissions:
actions: read
contents: read
issues: write
pull-requests: write
statuses: write
env:
FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
CODE_MOWER_GATE_CONTEXT: code-mower/gate
CODE_MOWER_CONTEXT_REQUIRED: "false"
CODE_MOWER_GATE_ENABLE_AUTO_MERGE: "true"
CODE_MOWER_OWNER_LABEL: "needs-owner"
CODE_MOWER_OWNER_SITTING_LABEL: "owner-sitting"
CODE_MOWER_OWNER_LOGIN: ""
CODE_MOWER_OWNER_LOGIN_OVERRIDE: ${{ vars.CODE_MOWER_OWNER_LOGIN || '' }}
CODE_MOWER_GATE_OVERRIDE_LABEL: "gate:override"
CODE_MOWER_DECISION_AUTHORITIES: ""
CODE_MOWER_DECISION_AUTHORITIES_OVERRIDE: ${{ vars.CODE_MOWER_DECISION_AUTHORITIES || '' }}
CODE_MOWER_AUTHOR_EXCLUSION_JSON: "{\"authors\":{\"chatgpt-codex-connector[bot]\":\"codex\",\"claude[bot]\":\"claude\",\"cursor[bot]\":\"cursor\",\"devin-ai-integration\":\"devin\",\"devin-ai-integration[bot]\":\"devin\",\"grok-bot[bot]\":\"cursor\"},\"branch_prefixes\":{\"cursor/\":\"cursor\",\"devin/\":\"devin\"},\"enabled\":true,\"labels\":{\"builder:claude\":\"claude\",\"builder:codex\":\"codex\",\"builder:cursor\":\"cursor\",\"builder:devin\":\"devin\",\"builder:gitar\":\"gitar\",\"builder:grok-bot\":\"cursor\",\"builder:muse\":\"muse\"},\"require_verified_lineage\":true}"
CODE_MOWER_GATE_LANES_JSON: "[{\"author_lane\":\"codex\",\"authors_env\":\"CODEX_BOT_AUTHORS\",\"blocked\":\"codex-audit-blocked\",\"bot_authors\":\"codex-audit-bot,codex-audit-bot[bot],github-actions[bot]\",\"builder_label\":\"builder:codex\",\"decision_coverage\":true,\"display_name\":\"Codex\",\"done\":\"codex-audit-done\",\"github_actions_workflows\":\".github/workflows/local-cli-audit.yml,.github/workflows/local-audit-publication.yml\",\"id\":\"codex\"},{\"author_lane\":\"claude\",\"authors_env\":\"CLAUDE_AUDIT_BOT_AUTHORS\",\"blocked\":\"claude-audit-blocked\",\"bot_authors\":\"claude-audit-bot,claude-audit-bot[bot],github-actions[bot]\",\"builder_label\":\"builder:claude\",\"decision_coverage\":true,\"display_name\":\"Claude\",\"done\":\"claude-audit-done\",\"github_actions_workflows\":\".github/workflows/local-cli-audit.yml,.github/workflows/local-audit-publication.yml\",\"id\":\"claude_audit\"}]"
jobs:
gate:
name: publish Code Mower gate status
if: >-
(github.event_name != 'pull_request_target' || github.event.pull_request.head.repo.full_name == github.repository) &&
(github.event_name != 'issue_comment' || (github.event.issue.pull_request && startsWith(github.event.comment.body, 'Code Mower context input') && contains(fromJSON('["OWNER","MEMBER","COLLABORATOR"]'), github.event.comment.author_association)))
runs-on: ubuntu-latest
timeout-minutes: 5
steps:
- name: Check out Code Mower support files from default branch
uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
with:
ref: ${{ github.event.repository.default_branch }}
persist-credentials: false
- name: Clear stale Code Mower gate override
if: >-
github.event_name == 'pull_request_target' &&
github.event.action == 'synchronize' &&
env.CODE_MOWER_GATE_OVERRIDE_LABEL != ''
env:
GH_TOKEN: ${{ github.token }}
PR_NUMBER: ${{ github.event.pull_request.number }}
run: |
set -euo pipefail
gh issue edit "${PR_NUMBER}" \
--repo "${GITHUB_REPOSITORY}" \
--remove-label "${CODE_MOWER_GATE_OVERRIDE_LABEL}" && exit 0
if gh api "repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/labels" \
--jq '.[].name' | grep -Fxq -- "${CODE_MOWER_GATE_OVERRIDE_LABEL}"; then
echo "::warning::Failed to remove stale ${CODE_MOWER_GATE_OVERRIDE_LABEL} label; publishing gate failure status."
echo "CODE_MOWER_GATE_OVERRIDE_CLEAR_FAILED=true" >> "${GITHUB_ENV}"
exit 0
fi
echo "::notice::${CODE_MOWER_GATE_OVERRIDE_LABEL} was already absent."
- name: Publish Code Mower gate status
env:
GH_TOKEN: ${{ github.token }}
CODE_MOWER_GATE_AUTOMERGE_TOKEN: ${{ secrets.CODE_MOWER_GATE_AUTOMERGE_TOKEN || secrets.DISPATCH_TOKEN || '' }}
PR_NUMBER: ${{ github.event.pull_request.number || github.event.issue.number || github.event.inputs.pr_number || github.event.workflow_run.pull_requests[0].number }}
HEAD_SHA: ${{ github.event.pull_request.head.sha || github.event.inputs.head_sha }}
WORKFLOW_RUN_HEAD_SHA: ${{ github.event.workflow_run.head_sha || '' }}
WORKFLOW_RUN_HEAD_BRANCH: ${{ github.event.workflow_run.head_branch || '' }}
WORKFLOW_RUN_HEAD_REPOSITORY: ${{ github.event.workflow_run.head_repository.full_name || '' }}
CLAUDE_AUDIT_BOT_AUTHORS: ${{ vars.CLAUDE_AUDIT_BOT_AUTHORS || '' }}
CODEX_BOT_AUTHORS: ${{ vars.CODEX_BOT_AUTHORS || '' }}
run: |
set -euo pipefail
if [ -z "${PR_NUMBER:-}" ] && [ -n "${WORKFLOW_RUN_HEAD_SHA:-}" ]; then
PR_NUMBER="$(
gh api -H "Accept: application/vnd.github+json" \
"repos/${GITHUB_REPOSITORY}/commits/${WORKFLOW_RUN_HEAD_SHA}/pulls?per_page=100" \
--jq 'map(select(.state == "open")) | .[0].number // ""' 2>/dev/null || true
)"
export PR_NUMBER
fi
if [ -z "${PR_NUMBER:-}" ] && [ -n "${WORKFLOW_RUN_HEAD_BRANCH:-}" ]; then
repo_owner="${GITHUB_REPOSITORY%%/*}"
head_owner="${repo_owner}"
if [ -n "${WORKFLOW_RUN_HEAD_REPOSITORY:-}" ]; then
head_owner="${WORKFLOW_RUN_HEAD_REPOSITORY%%/*}"
fi
PR_NUMBER="$(
gh api "repos/${GITHUB_REPOSITORY}/pulls" \
--method GET \
-f state=open \
-f "head=${head_owner}:${WORKFLOW_RUN_HEAD_BRANCH}" \
-F per_page=10 \
--jq '.[0].number // ""' 2>/dev/null || true
)"
export PR_NUMBER
fi
if [ -z "${PR_NUMBER:-}" ]; then
echo "::notice::No PR number available for Code Mower gate."
exit 0
fi
labels_file="$(mktemp)"
comments_file="$(mktemp)"
events_file="$(mktemp)"
pr_file="$(mktemp)"
workflow_runs_file="$(mktemp)"
audit_runs_file="$(mktemp)"
gh api "repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/labels" > "${labels_file}"
python3 - "${comments_file}" <<'PY'
import json
import os
import subprocess
import sys
try:
from tools.audit_labeler_lib import (
GITHUB_COMMENT_RESPONSE_BYTES,
GitHubCommentPage,
decode_github_response,
lineage_history,
)
except ImportError: # pragma: no cover - package fallback
from code_mower.audit_labeler_lib import (
GITHUB_COMMENT_RESPONSE_BYTES,
GitHubCommentPage,
decode_github_response,
lineage_history,
)
def fetch(page, size):
completed = subprocess.run(
["gh", "api", f"repos/{os.environ['GITHUB_REPOSITORY']}/issues/"
f"{os.environ['PR_NUMBER']}/comments?per_page={size}&page={page}"],
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
if completed.returncode != 0:
raise RuntimeError("authenticated GitHub comment request failed")
payload = decode_github_response(
completed.stdout, maximum_bytes=GITHUB_COMMENT_RESPONSE_BYTES)
return GitHubCommentPage(payload, len(completed.stdout))
try:
comments = lineage_history(fetch, return_raw=True)
result = [comments]
except Exception as exc:
result = {"code": "bounded_comment_history", "message": str(exc)}
with open(sys.argv[1], "w", encoding="utf-8") as handle:
json.dump(result, handle)
PY
gh api --paginate --slurp "repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/timeline?per_page=100" > "${events_file}"
gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}" > "${pr_file}"
if [ -z "${HEAD_SHA:-}" ]; then
HEAD_SHA="$(python3 - "${pr_file}" <<'PY'
import json
import sys
with open(sys.argv[1], encoding="utf-8") as handle:
payload = json.load(handle)
print(str(((payload.get("head") or {}).get("sha")) or ""))
PY
)"
export HEAD_SHA
fi
if [ -z "${HEAD_SHA:-}" ]; then
echo "::notice::No PR head SHA available for Code Mower gate."
exit 0
fi
if python3 - "${workflow_runs_file}" <<'PY'
import json
import os
import subprocess
import sys
try:
from tools.audit_labeler_lib import AUDIT_RUN_NON_TERMINAL_STATUSES
except ImportError: # pragma: no cover - package fallback
from code_mower.audit_labeler_lib import AUDIT_RUN_NON_TERMINAL_STATUSES
pages = []
repo = os.environ["GITHUB_REPOSITORY"]
for status in sorted(AUDIT_RUN_NON_TERMINAL_STATUSES):
completed = subprocess.run(
[
"gh",
"api",
"--paginate",
"--slurp",
f"repos/{repo}/actions/runs?event=pull_request_target&status={status}&per_page=100",
],
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
if completed.returncode != 0:
raise SystemExit(1)
try:
payload = json.loads(completed.stdout or "[]")
except json.JSONDecodeError:
raise SystemExit(1)
if isinstance(payload, list):
pages.extend(payload)
elif isinstance(payload, dict):
pages.append(payload)
with open(sys.argv[1], "w", encoding="utf-8") as handle:
json.dump(pages, handle)
PY
then
python3 - "${workflow_runs_file}" "${audit_runs_file}" <<'PY'
import json
import os
import subprocess
import sys
try:
from tools.audit_labeler_lib import (
AUDIT_RUN_NON_TERMINAL_STATUSES,
audit_run_matches_pr_head,
audit_run_workflow_path,
workflow_path_matches,
)
except ImportError: # pragma: no cover - package fallback
from code_mower.audit_labeler_lib import (
AUDIT_RUN_NON_TERMINAL_STATUSES,
audit_run_matches_pr_head,
audit_run_workflow_path,
workflow_path_matches,
)
def trusted_workflows_for_lanes():
values = set()
for lane in lanes:
raw = str(lane.get("github_actions_workflows") or "")
values.update(item.strip() for item in raw.split(",") if item.strip())
return values
def workflow_runs_from_payload(payload):
if isinstance(payload, dict):
return payload.get("workflow_runs", [])
runs = []
if isinstance(payload, list):
for page in payload:
if isinstance(page, dict):
page_runs = page.get("workflow_runs", [])
if isinstance(page_runs, list):
runs.extend(page_runs)
return runs
with open(sys.argv[1], encoding="utf-8") as handle:
workflow_runs_payload = json.load(handle)
lanes = json.loads(os.environ.get("CODE_MOWER_GATE_LANES_JSON", "[]"))
lanes = [lane for lane in lanes if isinstance(lane, dict)]
trusted_workflows = trusted_workflows_for_lanes()
head_sha = os.environ.get("HEAD_SHA", "")
pr_number = str(os.environ.get("PR_NUMBER", ""))
runs = workflow_runs_from_payload(workflow_runs_payload)
entries = []
for run in runs:
if not isinstance(run, dict):
continue
if str(run.get("status") or "") not in AUDIT_RUN_NON_TERMINAL_STATUSES:
continue
if not workflow_path_matches(audit_run_workflow_path(run), trusted_workflows):
continue
if not audit_run_matches_pr_head(run, head_sha=head_sha, pr_number=pr_number):
continue
run_id = str(run.get("id") or "")
jobs = []
jobs_fetch_failed = False
if run_id:
completed = subprocess.run(
["gh", "api", f"repos/{os.environ['GITHUB_REPOSITORY']}/actions/runs/{run_id}/jobs?per_page=100"],
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
if completed.returncode == 0:
try:
jobs_payload = json.loads(completed.stdout or "{}")
raw_jobs = jobs_payload.get("jobs", []) if isinstance(jobs_payload, dict) else []
jobs = [job for job in raw_jobs if isinstance(job, dict)]
except json.JSONDecodeError:
jobs_fetch_failed = True
else:
jobs_fetch_failed = True
entries.append({"run": run, "jobs": jobs, "jobs_fetch_failed": jobs_fetch_failed})
with open(sys.argv[2], "w", encoding="utf-8") as handle:
json.dump(entries, handle)
PY
else
printf '[]' > "${audit_runs_file}"
fi
eval "$(
python3 - "${labels_file}" "${comments_file}" "${events_file}" "${pr_file}" "${audit_runs_file}" <<'PY'
import json
import os
import shlex
import sys
try:
from tools.context_review import required_for_checkout
from tools.audit_labeler_lib import (
GitHubToken,
attested_non_current_audit_heads,
audit_run_in_flight_detail_for_lane,
audit_verdict_newer_than_in_flight,
flatten_paginated_items,
lineage_core,
lineage_decision,
lineage_snapshot,
github_actions_comment_attested,
latest_current_audit_verdict,
latest_current_audit_verdict_detail,
)
except ImportError: # pragma: no cover - unit-test/package fallback
from code_mower.context_review import required_for_checkout
from code_mower.audit_labeler_lib import (
GitHubToken,
attested_non_current_audit_heads,
audit_run_in_flight_detail_for_lane,
audit_verdict_newer_than_in_flight,
flatten_paginated_items,
lineage_core,
lineage_decision,
lineage_snapshot,
github_actions_comment_attested,
latest_current_audit_verdict,
latest_current_audit_verdict_detail,
)
# github_actions_comment_attested verifies the hidden CODE_MOWER_AUDIT_RUN marker
# and requires comment_id/body_sha256 to match the issue comment being evaluated.
def emit(state, description):
raw_description = str(description).replace("\n", " ")
description = raw_description[:140]
print("gate_state=" + shlex.quote(state))
print("gate_description=" + shlex.quote(description))
print("gate_log_description=" + shlex.quote(raw_description))
try:
lanes = json.loads(os.environ.get("CODE_MOWER_GATE_LANES_JSON", "[]"))
exclusion = lineage_core.Identity.from_text(os.environ["CODE_MOWER_AUTHOR_EXCLUSION_JSON"])
with open(sys.argv[1], encoding="utf-8") as labels_handle:
labels_payload = json.load(labels_handle)
with open(sys.argv[2], encoding="utf-8") as comments_handle:
comments_payload = lineage_core._json(comments_handle.read())
with open(sys.argv[3], encoding="utf-8") as events_handle:
events_payload = json.load(events_handle)
with open(sys.argv[4], encoding="utf-8") as pr_handle:
pr_payload = lineage_core._json(pr_handle.read())
with open(sys.argv[5], encoding="utf-8") as audit_runs_handle:
audit_runs_payload = json.load(audit_runs_handle)
except Exception:
emit("failure", "Code Mower gate config could not be parsed")
raise SystemExit(0)
head_sha = os.environ.get("HEAD_SHA", "")
actual_head_sha = str(((pr_payload.get("head") or {}).get("sha")) or "")
if not head_sha or head_sha != actual_head_sha:
emit("failure", "workflow head_sha does not match PR head")
raise SystemExit(0)
labels = {
str(item.get("name") or "")
for item in labels_payload
if isinstance(item, dict)
}
lanes = [lane for lane in lanes if isinstance(lane, dict)]
try:
history = lineage_core.History.from_pages(comments_payload)
except ValueError:
detail = (comments_payload.get("message")
if isinstance(comments_payload, dict)
and comments_payload.get("code") == "bounded_comment_history"
else "lineage history unreadable")
emit("failure", "Code Mower " + str(detail))
raise SystemExit(0)
comments = flatten_paginated_items(comments_payload)
events = flatten_paginated_items(events_payload)
audit_runs = audit_runs_payload if isinstance(audit_runs_payload, list) else []
configured_decision_authorities = (
os.environ.get("CODE_MOWER_DECISION_AUTHORITIES_OVERRIDE", "").strip()
or os.environ.get("CODE_MOWER_DECISION_AUTHORITIES", "")
)
decision_authorities = {
item.strip()
for item in configured_decision_authorities.split(",")
if item.strip()
}
owner_label = os.environ.get("CODE_MOWER_OWNER_LABEL", "needs-owner")
owner_sitting_label = os.environ.get("CODE_MOWER_OWNER_SITTING_LABEL", "owner-sitting")
gate_override_label = os.environ.get("CODE_MOWER_GATE_OVERRIDE_LABEL", "gate:override")
override_clear_failed = (
os.environ.get("CODE_MOWER_GATE_OVERRIDE_CLEAR_FAILED", "").lower()
== "true"
)
owner_login = (
os.environ.get("CODE_MOWER_OWNER_LOGIN_OVERRIDE", "").strip()
or os.environ.get("CODE_MOWER_OWNER_LOGIN", "").strip()
)
repo_owner = os.environ["GITHUB_REPOSITORY"].split("/", 1)[0]
configured_override_owner = (
owner_login
if owner_login and owner_login.lower() != "todo_owner_login"
else repo_owner
)
override_owner = configured_override_owner.lower()
pr_number = str(pr_payload.get("number") or os.environ.get("PR_NUMBER") or "")
pr_author = str(((pr_payload.get("user") or {}).get("login") or ""))
github_tokens = (GitHubToken("GH_TOKEN", os.environ.get("GH_TOKEN", "")),)
def mapping(section):
value = exclusion.get(section) if isinstance(exclusion, dict) else {}
return value if isinstance(value, dict) else {}
def trusted_authors(lane):
raw = str(lane.get("bot_authors") or "")
env_raw = os.environ.get(str(lane.get("authors_env") or ""), "")
merged = raw + "," + env_raw
return {item.strip().lower() for item in merged.split(",") if item.strip()}
def trusted_github_actions_workflows(lane):
raw = str(lane.get("github_actions_workflows") or "")
return {item.strip() for item in raw.split(",") if item.strip()}
def trusted_comment_author(lane, author, body, comment_id, expected_head_sha=None):
authors = trusted_authors(lane)
author_login = author.strip().lower()
if author_login not in authors:
return False
if author_login != "github-actions[bot]":
return True
attested_head_sha = expected_head_sha or head_sha
return github_actions_comment_attested(
repo=os.environ["GITHUB_REPOSITORY"],
body=body,
comment_id=comment_id,
issue_number=int(pr_number),
head_sha=attested_head_sha,
workflow_paths=trusted_github_actions_workflows(lane),
tokens=github_tokens,
)
context_required = required_for_checkout(
'code-mower.yml', fallback=os.environ.get('CODE_MOWER_CONTEXT_REQUIRED') == 'true'
)
def latest_current_verdict(lane):
return latest_current_audit_verdict(
lane,
comments,
head_sha=head_sha,
trusted_comment_author=trusted_comment_author,
decision_authorities=tuple(decision_authorities),
context_required=context_required,
)
def latest_current_verdict_detail(lane):
return latest_current_audit_verdict_detail(
lane,
comments,
head_sha=head_sha,
trusted_comment_author=trusted_comment_author,
decision_authorities=tuple(decision_authorities),
context_required=context_required,
)
try:
target, observed_author, observed_labels = lineage_snapshot(
os.environ["GITHUB_REPOSITORY"], int(pr_number), pr_payload)
if set(observed_labels) != labels:
raise lineage_core.ContractError("Fetched label snapshots differ")
_, lineage = lineage_decision(target, exclusion,
lineage_core.Authorities(decision_authorities), history,
author=observed_author, labels=observed_labels)
except (ValueError, KeyError, TypeError):
emit("failure", "Code Mower lineage contract unreadable; owner action required")
raise SystemExit(0)
if lineage.status != "ready":
emit("failure" if lineage.status == "conflict" else "pending", "lineage " + lineage.reason)
raise SystemExit(0)
builder_matches = list(lineage.contributors)
def lane_display(lane):
return str(lane.get("display_name") or lane.get("id") or lane.get("done") or "")
def in_flight_lanes(required):
names = []
for lane in required:
verdict = latest_current_verdict_detail(lane)
for entry in audit_runs:
if not isinstance(entry, dict):
continue
detail = audit_run_in_flight_detail_for_lane(
entry,
lane,
head_sha=head_sha,
pr_number=pr_number,
)
if detail is None:
continue
if audit_verdict_newer_than_in_flight(verdict, detail):
continue
names.append(
lane_display(lane) + " (" + detail.description() + ")"
)
break
return names
blocked = [
lane_display(lane)
for lane in lanes
if latest_current_verdict(lane) == "blocked"
]
override_actor = ""
override_event_key = None
head_move_event_key = None
def timeline_sha(value):
if isinstance(value, dict):
return str(value.get("sha") or value.get("id") or "")
return str(value or "")
def event_key(index, event):
created = str(event.get("created_at") or "")
return (created, index)
def is_current_head_move(event):
event_name = str(event.get("event") or "")
if event_name == "committed":
return str(event.get("sha") or "") == head_sha
if event_name == "head_ref_force_pushed":
return (
timeline_sha(event.get("after_commit")) == head_sha
or str(event.get("commit_id") or "") == head_sha
)
if event_name == "head_ref_restored":
return str(event.get("commit_id") or "") == head_sha
return False
if gate_override_label:
for index, event in enumerate(events):
if is_current_head_move(event):
head_move_event_key = event_key(index, event)
if event.get("event") != "labeled":
continue
if str(((event.get("label") or {}).get("name")) or "") != gate_override_label:
continue
override_actor = str(((event.get("actor") or {}).get("login")) or "").lower()
override_event_key = event_key(index, event)
override_current_head = (
override_event_key is not None
and (
head_move_event_key is None
or override_event_key > head_move_event_key
)
)
print(
"history_warning_heads="
+ shlex.quote(
json.dumps(
attested_non_current_audit_heads(
lanes,
comments,
head_sha=head_sha,
trusted_comment_author=trusted_comment_author,
)
)
)
)
if owner_label in labels:
emit("pending", owner_label + ": waiting on owner")
elif owner_sitting_label and owner_sitting_label in labels:
emit("pending", owner_sitting_label + ": waiting on owner")
elif gate_override_label and gate_override_label in labels and override_actor != override_owner:
emit("failure", gate_override_label + " was not owner-applied")
elif gate_override_label and gate_override_label in labels and override_clear_failed:
emit("failure", gate_override_label + " could not be cleared after head change")
elif gate_override_label and gate_override_label in labels and not override_current_head:
emit("failure", gate_override_label + " is stale for current head")
elif gate_override_label and gate_override_label in labels:
emit("success", "owner gate override")
elif blocked:
emit("failure", "blocked audit: " + ", ".join(blocked))
else:
excluded_builder = builder_matches[0] if builder_matches else ""
required = [
lane
for lane in lanes
if lineage_core.admit(lineage, str(lane.get("author_lane") or lane.get("id") or ""))
]
in_flight = in_flight_lanes(required)
missing = [
lane_display(lane)
for lane in required
if str(lane.get("done") or "") not in labels
or latest_current_verdict(lane) != "done"
]
if not required:
if excluded_builder:
emit("failure", "no independent Code Mower audit lane remains")
else:
emit("failure", "no Code Mower merge-authority lanes configured")
elif in_flight:
emit("pending", "audit in flight: " + ", ".join(in_flight))
elif missing:
emit("pending", "waiting for audit: " + ", ".join(missing))
else:
emit("success", "Code Mower merge gate passed")
PY
)"
history_warning_file="$(mktemp)"
python3 - "${history_warning_heads:-[]}" "${comments_file}" "${history_warning_file}" <<'PY'
import json
import os
import subprocess
import sys
try:
heads = json.loads(sys.argv[1])
except json.JSONDecodeError:
heads = []
heads = [str(head) for head in heads if str(head)]
try:
with open(sys.argv[2], encoding="utf-8") as handle:
comments_payload = json.load(handle)
except Exception:
comments_payload = []
comments = []
for page in comments_payload:
if isinstance(page, list):
comments.extend(item for item in page if isinstance(item, dict))
head_sha = os.environ.get("HEAD_SHA", "")
repo = os.environ.get("GITHUB_REPOSITORY", "")
for previous in heads:
marker = f"<!-- CODE_MOWER_HISTORY_REWRITE_WARNING previous={previous} current={head_sha} -->"
if any(marker in str(comment.get("body") or "") for comment in comments):
continue
completed = subprocess.run(
["gh", "api", f"repos/{repo}/compare/{previous}...{head_sha}", "--jq", ".status"],
check=False,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
text=True,
)
if completed.returncode != 0:
continue
status = completed.stdout.strip()
if status in {"ahead", "identical"}:
continue
body = (
marker
+ "\nCode Mower notice: previously audited head `"
+ previous[:12]
+ "` is no longer an ancestor of current head `"
+ head_sha[:12]
+ "`; commits may have been dropped. Only current-head audit verdicts count."
)
with open(sys.argv[3], "w", encoding="utf-8") as handle:
json.dump({"body": body}, handle)
break
PY
if [ -s "${history_warning_file}" ]; then
gh api -X POST "repos/${GITHUB_REPOSITORY}/issues/${PR_NUMBER}/comments" \
--input "${history_warning_file}" >/dev/null || {
echo "::notice::Could not post Code Mower history-rewrite warning."
}
fi
target_url="https://github.com/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
gh api -X POST "repos/${GITHUB_REPOSITORY}/statuses/${HEAD_SHA}" \
-f state="${gate_state}" \
-f context="${CODE_MOWER_GATE_CONTEXT}" \
-f description="${gate_description}" \
-f target_url="${target_url}" >/dev/null
echo "Code Mower gate: ${gate_state} - ${gate_log_description:-${gate_description}}"
# The merge signal is the commit status above. Keep expected
# pending/failure verdicts from creating a second required check-run.
if [ "${gate_state}" != "success" ] || [ "${CODE_MOWER_GATE_ENABLE_AUTO_MERGE}" != "true" ]; then
exit 0
fi
# One head-pinned merge path for both merge states: `gh pr merge
# --auto` queues auto-merge while required checks are still pending
# and merges promptly once the PR is mergeable, so queue-later and
# merge-now share a single call. --match-head-commit refuses a moved
# head instead of merging unaudited commits.
# CODE_MOWER_GATE_AUTOMERGE_BEGIN
automerge_token="${CODE_MOWER_GATE_AUTOMERGE_TOKEN:-${GH_TOKEN:-}}"
automerge_attempt=1
automerge_max_attempts="${CODE_MOWER_GATE_AUTOMERGE_MAX_ATTEMPTS:-5}"
automerge_delay="${CODE_MOWER_GATE_AUTOMERGE_RETRY_DELAY:-5}"
case "${automerge_max_attempts}" in
''|*[!0-9]*) automerge_max_attempts=5 ;;
esac
case "${automerge_delay}" in
''|*[!0-9]*) automerge_delay=5 ;;
esac
while true; do
automerge_stdout_file="$(mktemp)"
automerge_stderr_file="$(mktemp)"
if GH_TOKEN="${automerge_token}" gh pr merge "${PR_NUMBER}" \
--repo "${GITHUB_REPOSITORY}" \
--auto --squash --match-head-commit "${HEAD_SHA}" \
>"${automerge_stdout_file}" 2>"${automerge_stderr_file}"; then
automerge_result="$(head -n 1 "${automerge_stdout_file}")"
rm -f "${automerge_stdout_file}" "${automerge_stderr_file}"
echo "::notice::Code Mower gate merge request accepted for head ${HEAD_SHA}: ${automerge_result:-merged or queued}."
break
fi
automerge_error="$(head -n 1 "${automerge_stderr_file}")"
rm -f "${automerge_stdout_file}" "${automerge_stderr_file}"
case "${automerge_error}" in
*[Uu]nstable*|*[Rr]ate\ limit*|*try\ again*|*timed\ out*|*[Tt]imeout*|*deadline\ exceeded*|*HTTP\ 502*|*HTTP\ 503*|*HTTP\ 504*|*Bad\ gateway*|*Service\ unavailable*)
automerge_retry=true
;;
*)
automerge_retry=false
;;
esac
if [ "${automerge_retry}" != "true" ]; then
echo "::notice::Code Mower gate is green, but auto-merge was refused: $(printf '%.200s' "${automerge_error:-unknown error}"). Confirm token permissions, branch policy, and PR head."
break
fi
if [ "${automerge_attempt}" -ge "${automerge_max_attempts}" ]; then
echo "::notice::Code Mower gate is green, but auto-merge stayed transiently unavailable after ${automerge_attempt} attempts: $(printf '%.200s' "${automerge_error:-unknown error}"). A later gate run or manual merge can complete it."
break
fi
automerge_attempt="$((automerge_attempt + 1))"
sleep "${automerge_delay}"
automerge_delay="$((automerge_delay * 2))"
done
# CODE_MOWER_GATE_AUTOMERGE_END