code-mower-gate.yml

# 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