fix(sop-checklist): implement /sop-n/a N/A declarations + review-check 403 fix
Some checks are pending
Block internal-flavored paths / Block forbidden paths (pull_request) Waiting to run
CI / Detect changes (pull_request) Waiting to run
CI / Platform (Go) (pull_request) Waiting to run
CI / Canvas (Next.js) (pull_request) Waiting to run
CI / Shellcheck (E2E scripts) (pull_request) Waiting to run
CI / Canvas Deploy Reminder (pull_request) Blocked by required conditions
CI / Python Lint & Test (pull_request) Waiting to run
CI / all-required (pull_request) Blocked by required conditions
E2E API Smoke Test / detect-changes (pull_request) Waiting to run
E2E API Smoke Test / E2E API Smoke Test (pull_request) Blocked by required conditions
E2E Chat / detect-changes (pull_request) Waiting to run
E2E Chat / E2E Chat (pull_request) Blocked by required conditions
E2E Staging Canvas (Playwright) / detect-changes (pull_request) Waiting to run
E2E Staging Canvas (Playwright) / Canvas tabs E2E (pull_request) Blocked by required conditions
E2E Staging External Runtime / E2E Staging External Runtime (pull_request) Waiting to run
E2E Staging SaaS (full lifecycle) / pr-validate (pull_request) Waiting to run
E2E Staging SaaS (full lifecycle) / E2E Staging SaaS (pull_request) Waiting to run
Handlers Postgres Integration / detect-changes (pull_request) Waiting to run
Handlers Postgres Integration / Handlers Postgres Integration (pull_request) Blocked by required conditions
Harness Replays / detect-changes (pull_request) Waiting to run
Harness Replays / Harness Replays (pull_request) Blocked by required conditions
lint-continue-on-error-tracking / lint-continue-on-error-tracking (pull_request) Waiting to run
lint-mask-pr-atomicity / lint-mask-pr-atomicity (pull_request) Waiting to run
Lint pre-flip continue-on-error / Verify continue-on-error flips have run-log proof (pull_request) Waiting to run
lint-required-context-exists-in-bp / lint-required-context-exists-in-bp (pull_request) Waiting to run
lint-required-no-paths / lint-required-no-paths (pull_request) Waiting to run
Lint workflow YAML (Gitea-1.22.6-hostile shapes) / Lint workflow YAML for Gitea-1.22.6-hostile shapes (pull_request) Waiting to run
publish-runtime-autobump / pr-validate (pull_request) Waiting to run
publish-runtime-autobump / bump-and-tag (pull_request) Waiting to run
review-check-tests / review-check.sh regression tests (pull_request) Waiting to run
Runtime PR-Built Compatibility / detect-changes (pull_request) Waiting to run
Runtime PR-Built Compatibility / PR-built wheel + import smoke (pull_request) Blocked by required conditions
Secret scan / Scan diff for credential-shaped strings (pull_request) Waiting to run
Ops Scripts Tests / Ops scripts (unittest) (pull_request) Waiting to run
gate-check-v3 / gate-check (pull_request) Waiting to run
qa-review / approved (pull_request) Waiting to run
security-review / approved (pull_request) Waiting to run
sop-checklist / all-items-acked (pull_request) Waiting to run
sop-tier-check / tier-check (pull_request) Waiting to run
audit-force-merge / audit (pull_request) Waiting to run

Cherry-pick of infra/main-sop-na-fix N/A implementation + follow-up fixes.

N/A gate implementation (mc#1233 follow-up):
  - Adds separate _NA_DIRECTIVE_RE for /sop-n/a <gate> [reason] parsing.
  - parse_directives() now returns (directives, na_directives) tuple.
  - compute_na_state() evaluates N/A declarations per gate with
    team-membership probe (same 403 semantics as acks).
  - Posts 'sop-checklist / na-declarations (pull_request)' status
    with state=success when a gate is validly declared N/A by a
    non-author team member; state=failure otherwise.
  - review-check.sh reads this status to waive qa-review/security-review
    Gitea-APPROVE requirement when N/A is declared.

review-check 403 follow-up:
  - Team-membership API returns 403 if token owner is not in the team.
    Changed from exit 1 (hard-fail entire gate) to continue (skip this
    candidate, check others). Prevents a single 403 from blocking a gate
    when other valid team-members exist.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Molecule AI · core-devops 2026-05-15 22:38:19 +00:00
parent 7b8b9dbf18
commit ffd525061e
2 changed files with 213 additions and 16 deletions

View File

@ -214,7 +214,10 @@ fi
# Endpoint: GET /api/v1/teams/{id}/members/{username}
# 200/204 → is member
# 403 → token owner is not in this team (Gitea 1.22.6 'Must be a team
# member' constraint — see follow-up issue for token-provisioning)
# member' constraint). The evaluator skips this candidate and
# continues to check others. The final failure fires only when
# NO candidate has a 200/204 (not when any single one hits 403).
# See RFC#324 token-scope follow-up issue for long-term fix.
# 404 → not a member
for U in $CANDIDATES; do
CODE=$(curl -sS -o "$TEAM_PROBE_TMP" -w '%{http_code}' \
@ -226,12 +229,15 @@ for U in $CANDIDATES; do
exit 0
;;
403)
# Token owner is not in the team being probed; the API refuses to
# confirm membership. This is the RFC#324 follow-up token-scope gap.
# Fail closed — never grant approval on a 403; surface clearly.
echo "::error::team-probe for ${U} in ${TEAM} returned 403 (token owner not in ${TEAM} team — RFC#324 token-scope follow-up). Cannot confirm membership; failing closed."
# Token owner is not in the team being probed; Gitea 1.22.6 refuses
# to confirm membership in this case. Do NOT hard-fail the gate on a
# 403 — doing so would fail the entire gate if ANY candidate triggers
# a 403, even when other valid team-members exist. Instead skip this
# candidate and continue checking others. If all candidates produce
# 403 (token owner can't query any of them) the final exit fires.
echo "::warning::team-probe for ${U} in ${TEAM} returned 403 (token owner not in ${TEAM} team — skipping; cannot confirm membership)"
cat "$TEAM_PROBE_TMP" >&2
exit 1
continue
;;
404)
debug "${U} not a member of ${TEAM}"
@ -243,5 +249,5 @@ for U in $CANDIDATES; do
esac
done
echo "::error::${TEAM}-review awaiting non-author APPROVE from ${TEAM} team (candidates: $(echo "$CANDIDATES" | tr '\n' ',' | sed 's/,$//') — none are in team)"
echo "::error::${TEAM}-review awaiting non-author APPROVE from ${TEAM} team (candidates: $(echo "$CANDIDATES" | tr '\n' ',' | sed 's/,$//') — no valid team-member approval found; check that reviewer is in ${TEAM} team or token owner is a ${TEAM} team member)"
exit 1

View File

@ -102,7 +102,7 @@ def normalize_slug(raw: str, numeric_aliases: dict[int, str] | None = None) -> s
# ---------------------------------------------------------------------------
# Comment parsing — /sop-ack and /sop-revoke
# Comment parsing — /sop-ack, /sop-revoke, and /sop-n/a
# ---------------------------------------------------------------------------
# A directive must be on its own line. Permits leading whitespace.
@ -114,23 +114,33 @@ _DIRECTIVE_RE = re.compile(
re.MULTILINE,
)
# /sop-n/a <gate> [reason] — declare a qa/sec gate N/A.
# Gate names: qa-review, security-review (match review-check.sh context names).
_NA_DIRECTIVE_RE = re.compile(
r"^[ \t]*/sop-n/a[ \t]+([A-Za-z0-9_\-]+)(?:[ \t]+(.*))?[ \t]*$",
re.MULTILINE,
)
def parse_directives(
comment_body: str,
numeric_aliases: dict[int, str],
) -> tuple[list[tuple[str, str, str]], list]:
"""Extract /sop-ack and /sop-revoke directives from a comment body.
) -> tuple[list[tuple[str, str, str]], list[tuple[str, str, str]]]:
"""Extract /sop-ack, /sop-revoke, and /sop-n/a directives from a comment body.
Returns (directives, na_directives) where:
directives is a list of (kind, canonical_slug, note) tuples
kind is "sop-ack" or "sop-revoke"
canonical_slug is the normalized form (or "" if unparseable)
note is the trailing free-text (may be "")
na_directives is reserved for future N/A handling (always [] for now)
na_directives is a list of (gate_name, reason) tuples
gate_name is "qa-review" or "security-review" (raw from comment)
reason is the free-text after the gate name (may be "")
"""
out: list[tuple[str, str, str]] = []
na_out: list[tuple[str, str, str]] = []
if not comment_body:
return out, []
return out, na_out
for m in _DIRECTIVE_RE.finditer(comment_body):
kind = m.group(1)
raw_slug = (m.group(2) or "").strip()
@ -161,7 +171,11 @@ def parse_directives(
# If we collapsed multi-word slug into kebab and there's a
# trailing-text group too, append it.
out.append((kind, canonical, note_from_group))
return out, []
for m in _NA_DIRECTIVE_RE.finditer(comment_body):
gate_raw = (m.group(1) or "").strip()
reason = (m.group(2) or "").strip()
na_out.append((gate_raw.lower(), reason))
return out, na_out
# ---------------------------------------------------------------------------
@ -304,6 +318,82 @@ def compute_ack_state(
}
def compute_na_state(
comments: list[dict[str, Any]],
pr_author: str,
na_gates: dict[str, dict[str, Any]],
team_membership_probe: "callable[[str, list[str]], list[str]]",
) -> dict[str, dict[str, Any]]:
"""Compute per-gate N/A declaration state.
Each comment is processed in chronological order. The most-recent
N/A directive per (commenter, gate) wins.
Returns a dict keyed by gate name:
{
"qa-review": {
"declared": True,
"declared_by": "core-qa-agent",
"reason": "CI/non-security-touching",
"valid": True, # non-author + in required team
"error": None, # error string if invalid
},
...
}
Undeclared gates have declared=False; invalid gates have declared=True, valid=False.
"""
# Step 1: collapse N/A directives per (commenter, gate) — most recent wins.
latest_na: dict[tuple[str, str], tuple[str, str]] = {}
for c in comments:
body = c.get("body", "") or ""
user = (c.get("user") or {}).get("login", "")
if not user:
continue
_, na_directives = parse_directives(body, {})
for gate, reason in na_directives:
if gate not in na_gates:
continue
latest_na[(user, gate)] = (gate, reason)
# Step 2: initialise all gates as undeclared.
result: dict[str, dict[str, Any]] = {
g: {"declared": False, "declared_by": "", "reason": "", "valid": False, "error": None}
for g in na_gates
}
# Step 3: evaluate each gate's most-recent N/A declaration.
for (user, gate), (gate_name, reason) in latest_na.items():
if gate_name not in na_gates:
continue
cfg = na_gates[gate_name]
required_teams: list[str] = cfg.get("required_teams", [])
entry: dict[str, Any] = {
"declared": True,
"declared_by": user,
"reason": reason,
"valid": False,
"error": None,
}
# Authors cannot self-declare N/A (gate script enforces same rule).
if user == pr_author:
entry["error"] = "self-declare N/A rejected"
else:
# Probe team membership: is the declarer in any required team?
approved = team_membership_probe(f"na:{gate_name}", [user])
if user in approved:
entry["valid"] = True
else:
# 403 from team API means token owner not in that team.
# Fail-closed: treat unknown membership as invalid.
entry["error"] = f"{user} not in required team {required_teams}"
result[gate_name] = entry
return result
# ---------------------------------------------------------------------------
# Gitea API client
# ---------------------------------------------------------------------------
@ -463,10 +553,29 @@ def _load_config_minimal(path: str) -> dict[str, Any]:
tier_failure_mode), top-level list of maps (items:), and within an
item map: scalars + lists of scalars. Does NOT support nested lists,
YAML anchors, multi-doc, or flow style.
Key names containing '/' (e.g. n/a_gates) are handled by using
rpartition(':') splitting at the LAST colon so embedded colons
in the key are preserved.
"""
with open(path) as f:
lines = f.readlines()
return _parse_minimal_yaml(lines)
# Preprocess: for lines at indent 0 that contain '/' before ':',
# use rpartition so the key keeps the '/'. e.g.
# "n/a_gates:" → key="n/a_gates", val=""
# "n/a_gates: value" → key="n/a_gates", val="value"
processed: list[str] = []
for raw in lines:
stripped = raw.rstrip("\n")
indent = len(stripped) - len(stripped.lstrip(" "))
content = stripped.lstrip(" ")
if indent == 0 and "/" in content and ":" in content:
# Use rpartition so the last ':' is the key-value separator.
key, _, val = content.rpartition(":")
processed.append(" " * indent + key.strip() + ": " + val.strip())
else:
processed.append(stripped)
return _parse_minimal_yaml(processed)
def _parse_minimal_yaml(lines: list[str]) -> dict[str, Any]: # noqa: C901
@ -803,6 +912,90 @@ def main(argv: list[str] | None = None) -> int:
extra = " (" + "; ".join(extras) + ")" if extras else ""
print(f"::notice:: [WAIT] {slug} — no valid peer-ack yet{extra}")
# ----- N/A gate declarations (RFC#324 §N/A follow-up) -----
# sop-checklist.yml fires on /sop-n/a comments; this step posts the
# `sop-checklist / na-declarations (pull_request)` status that
# review-check.sh reads to waive the Gitea-APPROVE requirement.
na_gates: dict[str, Any] = cfg.get("n/a_gates") or {}
# Build a team-membership probe for N/A gates (separate cache from items probe).
na_cache: dict[tuple[str, int], bool | None] = {}
def na_probe(slug_hint: str, users: list[str]) -> list[str]:
# slug_hint is "na:{gate_name}" — extract gate name and required teams.
gate_name = slug_hint.removeprefix("na:")
gate_cfg = na_gates.get(gate_name, {})
team_names: list[str] = gate_cfg.get("required_teams", [])
# Resolve team names → ids.
team_ids: list[int] = []
for tn in team_names:
tid = client.resolve_team_id(args.owner, tn) # noqa: SLF001
if tid is None:
code, data = client._req( # noqa: SLF001
"GET", f"/orgs/{args.owner}/teams"
)
if code == 200 and isinstance(data, list):
for t in data:
if t.get("name") == tn:
tid = t.get("id")
client._team_id_cache[(args.owner, tn)] = tid # noqa: SLF001
break
if tid is not None:
team_ids.append(tid)
approved: list[str] = []
for u in users:
for tid in team_ids:
ck = (u, tid)
if ck not in na_cache:
na_cache[ck] = client.is_team_member(tid, u) # noqa: SLF001
res = na_cache[ck]
if res is True:
approved.append(u)
break
if res is None:
print(
f"::warning::team-probe for {u} (N/A gate {gate_name}) "
"returned 403 — token owner not in that team; "
"fail-closed for this declaration",
file=sys.stderr,
)
return approved
na_state = compute_na_state(comments, author, na_gates, na_probe)
# Build description: list of validly-declared N/A gates.
na_approved_gates = [
g for g, entry in na_state.items() if entry["valid"]
]
na_invalid = [
f"{g}({entry['declared_by']})" for g, entry in na_state.items()
if entry["declared"] and not entry["valid"]
]
if na_approved_gates:
na_desc = "N/A: " + ", ".join(na_approved_gates)
elif na_invalid:
na_desc = "invalid N/A: " + ", ".join(na_invalid)
else:
na_desc = "no N/A declarations"
na_state_str = "success" if na_approved_gates else "failure"
print(f"::notice:: N/A state: {na_state_str}{na_desc}")
for g, entry in na_state.items():
if entry["declared"]:
status_flag = "valid" if entry["valid"] else f"invalid: {entry['error']}"
print(f"::notice:: {g}: declared by {entry['declared_by']}{status_flag}")
target_url = f"https://{args.gitea_host}/{args.owner}/{args.repo}/pulls/{args.pr}"
if not args.dry_run:
na_context = "sop-checklist / na-declarations (pull_request)"
client.post_status(
args.owner, args.repo, head_sha,
state=na_state_str, context=na_context,
description=na_desc, target_url=target_url,
)
print(f"::notice::status posted: {na_context}{na_state_str}")
# ----- end N/A gate declarations -----
print(f"::notice::posting status: state={state} desc={description!r}")
if args.dry_run:
@ -810,8 +1003,6 @@ def main(argv: list[str] | None = None) -> int:
if args.exit_on_state:
return 0 if state in ("success", "pending") else 1
return 0
target_url = f"https://{args.gitea_host}/{args.owner}/{args.repo}/pulls/{args.pr}"
client.post_status(
args.owner, args.repo, head_sha,
state=state, context=args.status_context,