.gitea/scripts/main-red-watchdog.py

@@ -576,12 +576,14 @@ def run_once(*, dry_run: bool = False) -> int:
               f"{len(failed)} failed context(s)")
         file_or_update_red(sha, failed, debug, dry_run=dry_run)
     else:
-        # Green (or pending — pending is treated as not-red so we don't
-        # spam during the post-merge CI window). Close any stale issues
-        # from earlier SHAs only when we're actually green; pending
-        # means CI hasn't finished and the prior issue might still be
-        # accurate.
-        if status.get("state") == "success":
+        if status.get("state") in ("success", "pending"):
+            # Close stale main-red issues when main has no failures.
+            # `pending` is included because Gitea combined-state can stay
+            # `pending` even when all observable individual statuses are
+            # successful (some jobs still running). The `is_red()` check
+            # already confirmed 0 failures — closing on `pending` prevents
+            # stale issues from persisting across cron ticks while
+            # long-running jobs finish.
             closed = close_open_red_issues_for_other_shas(sha, dry_run=dry_run)
             if closed:
                 emit_loki_event(
@@ -589,8 +591,10 @@ def run_once(*, dry_run: bool = False) -> int:
                     [],
                 )
             print(f"::notice::main is GREEN at {sha[:10]} on {WATCH_BRANCH} "
-                  f"(closed {closed} stale issue(s))")
+                  f"(closed {closed} stale issue(s), combined={status.get('state')})")
         else:
+            print(f"::notice::main is RED/ERROR at {sha[:10]} on {WATCH_BRANCH} "
+                  f"(combined state={status.get('state')!r})")
             print(f"::notice::main is PENDING at {sha[:10]} on {WATCH_BRANCH} "
                   f"(combined state={status.get('state')!r}; no action)")
     return 0

.gitea/scripts/sop-checklist.py

@@ -68,7 +68,7 @@ import sys
 import urllib.error
 import urllib.parse
 import urllib.request
-from typing import Any, Callable
+from typing import Any
 
 
 # ---------------------------------------------------------------------------
@@ -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.
@@ -113,8 +113,7 @@ _DIRECTIVE_RE = re.compile(
     r"^[ \t]*/(sop-ack|sop-revoke|sop-n/a)[ \t]+([A-Za-z0-9_\- ]+?)(?:[ \t]+(.*))?[ \t]*$",
     re.MULTILINE,
 )
-
-
+# /sop-n/a <gate> [reason] — declare a qa/sec gate N/A.
 def parse_directives(
     comment_body: str,
     numeric_aliases: dict[int, str],
@@ -349,59 +348,78 @@ def compute_ack_state(
     }
 
 
-# ---------------------------------------------------------------------------
-# N/A-gate evaluation
-# ---------------------------------------------------------------------------
-
-
 def compute_na_state(
     comments: list[dict[str, Any]],
-    author: str,
-    na_gates: dict[str, Any],
-    probe: Callable[[str, list[str]], list[str]],
+    pr_author: str,
+    na_gates: dict[str, dict[str, Any]],
+    team_membership_probe: "callable[[str, list[str]], list[str]]",
 ) -> dict[str, dict[str, Any]]:
-    """Evaluate which N/A gates have a valid declaration from a team member.
+    """Compute per-gate N/A declaration state.
 
-    Returns dict[gate_name, dict] where each dict has:
-      declared: bool — at least one valid non-author team-member declared N/A
-      decl_ackers: list[str] — usernames who declared this gate N/A
-      rejected: dict with keys:
-        not_in_team: list[str] — users who tried but aren't in required teams
+    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.
     """
-    # Build per-user latest N/A directive (most-recent wins per RFC#324).
-    latest_na: dict[str, tuple[str, str]] = {}  # user → (gate, note)
+    # 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
-        for kind, gate, note in parse_directives(body, {})[1]:
-            # [1] = na_directives only
-            if gate in na_gates:
-                latest_na[user] = (gate, note)
-
-    result: dict[str, dict[str, Any]] = {}
-    for gate, gate_cfg in na_gates.items():
-        result[gate] = {
-            "declared": False,
-            "decl_ackers": [],
-            "rejected": {"not_in_team": []},
-        }
-        decl_ackers: list[str] = []
-        not_in_team: list[str] = []
-        for user, (g, _note) in latest_na.items():
-            if g != gate:
+        _, na_directives = parse_directives(body, {})
+        for _kind, gate, reason in na_directives:
+            if gate not in na_gates:
                 continue
-            if user == author:
-                continue  # authors cannot self-declare N/A
-            approved = probe(gate, [user])
-            if approved:
-                decl_ackers.append(user)
+            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:
-                not_in_team.append(user)
-        result[gate]["declared"] = bool(decl_ackers)
-        result[gate]["decl_ackers"] = decl_ackers
-        result[gate]["rejected"]["not_in_team"] = not_in_team
+                # 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
 
@@ -906,6 +924,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}")
 
+    target_url = f"https://{args.gitea_host}/{args.owner}/{args.repo}/pulls/{args.pr}"
+
+    # ----- 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}")
+
+    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:
@@ -913,8 +1015,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,
@@ -922,45 +1022,6 @@ def main(argv: list[str] | None = None) -> int:
     )
     print(f"::notice::status posted: {args.status_context} → {state}")
 
-    # --- N/A gate status (RFC#324 §N/A follow-up) ---
-    # Post a separate status so review-check.sh can discover N/A declarations
-    # and waive the Gitea-approve requirement for that gate.
-    na_state: dict[str, dict[str, Any]] = {}
-    if na_gates:
-        na_state = compute_na_state(comments, author, na_gates, probe)
-
-        na_descs: list[str] = []
-        for gate, s in na_state.items():
-            if s["declared"]:
-                na_descs.append(gate)
-            decl = s["decl_ackers"]
-            rej = s["rejected"]["not_in_team"]
-            if decl:
-                print(f"::notice::  [N/A OK] {gate} — declared by {','.join(decl)}")
-            if rej:
-                print(
-                    f"::notice::  [N/A REJ] {gate} — not-in-team: {','.join(rej)}",
-                    file=sys.stderr,
-                )
-
-        na_desc = ", ".join(sorted(na_descs)) if na_descs else "(none)"
-        na_status_state = "success" if na_descs else "pending"
-        # review-check.sh reads the description to discover which gates are N/A.
-        # Include the gate names so it can grep for them.
-        na_description = f"N/A: {na_desc}" if na_descs else "N/A: (none)"
-
-        if not args.dry_run:
-            client.post_status(
-                args.owner, args.repo, head_sha,
-                state=na_status_state,
-                context="sop-checklist / na-declarations (pull_request)",
-                description=na_description,
-                target_url=target_url,
-            )
-            print(
-                f"::notice::na-declarations status → {na_status_state}: {na_description}"
-            )
-
     # By default exit 0 — the POSTed status IS the gate, NOT the job
     # conclusion. If the job exits 1 BP will see TWO failure signals
     # (one from the job's auto-status, one from our POST), making the

.gitea/scripts/tests/_review_refire_fixture.py

@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""Stub Gitea API for review-refire-status.sh test scenarios.
+
+Reads $FIXTURE_STATE_DIR/scenario to decide what to return for each
+endpoint the review-refire-status.sh script calls (and review-check.sh
+which it invokes inline).  Also reads $FIXTURE_STATE_DIR/review_check_rc
+to control what review-check.sh exits with (PASS → 0, FAIL → 1).
+
+Scenarios:
+  open    — PR is open; review-check.sh runs and exits based on review_check_rc
+  closed  — PR is closed; script exits 0 with no-op
+
+review_check_rc file content:
+  PASS → review-check.sh exits 0 (success)
+  FAIL → review-check.sh exits 1 (failure)
+
+Usage:
+  FIXTURE_STATE_DIR=/tmp/x python3 _review_refire_fixture.py 8080
+"""
+
+import http.server
+import json
+import os
+import re
+import sys
+import urllib.parse
+
+
+STATE_DIR = os.environ.get("FIXTURE_STATE_DIR", "/tmp")
+
+
+def scenario() -> str:
+    p = os.path.join(STATE_DIR, "scenario")
+    if not os.path.isfile(p):
+        return "open"
+    with open(p).read() as f:
+        return f.read().strip()
+
+
+def review_check_rc() -> int:
+    p = os.path.join(STATE_DIR, "review_check_rc")
+    if os.path.isfile(p):
+        content = open(p).read().strip()
+        return 0 if content == "PASS" else 1
+    return 0  # default: pass
+
+
+class Handler(http.server.BaseHTTPRequestHandler):
+    def log_message(self, *args, **kwargs):
+        pass  # keep stdout quiet
+
+    def _json(self, code: int, body: dict) -> None:
+        payload = json.dumps(body).encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(payload)))
+        self.end_headers()
+        self.wfile.write(payload)
+
+    def _empty(self, code: int) -> None:
+        self.send_response(code)
+        self.send_header("Content-Length", "0")
+        self.end_headers()
+
+    def do_GET(self):
+        u = urllib.parse.urlparse(self.path)
+        path = u.path
+        sc = scenario()
+
+        # GET /repos/{owner}/{repo}/pulls/{pr_number}
+        m = re.match(r"^/api/v1/repos/([^/]+)/([^/]+)/pulls/(\d+)$", path)
+        if m:
+            return self._json(200, {
+                "number": int(m.group(3)),
+                "state": "closed" if sc == "closed" else "open",
+                "head": {"sha": "deadbeef0000111122223333444455556666"},
+                "base": {"ref": "main"},
+                "user": {"login": "feature-author"},
+            })
+
+        # GET /repos/{owner}/{repo}/pulls/{pr_number}/reviews
+        m = re.match(r"^/api/v1/repos/([^/]+)/([^/]+)/pulls/(\d+)/reviews$", path)
+        if m:
+            return self._json(200, [
+                {"state": "APPROVED", "dismissed": False,
+                 "user": {"login": "qa-member"}, "commit_id": "abc1234"},
+            ])
+
+        # GET /teams/{team_id}/members/{username}
+        m = re.match(r"^/api/v1/teams/(\d+)/members/([^/]+)$", path)
+        if m:
+            return self._empty(204)  # member
+
+        self._json(404, {"path": path})
+
+    def do_POST(self):
+        u = urllib.parse.urlparse(self.path)
+        path = u.path
+
+        # POST /repos/{owner}/{repo}/statuses/{sha}
+        m = re.match(r"^/api/v1/repos/([^/]+)/([^/]+)/statuses/([a-f0-9]+)$", path)
+        if m:
+            length = int(self.headers.get("Content-Length", 0))
+            body = self.rfile.read(length) if length else b"{}"
+            try:
+                data = json.loads(body)
+            except Exception:
+                data = {}
+            # Echo back what was posted so test can verify
+            return self._json(200, {"posted": data})
+
+        self._json(404, {"path": path})
+
+
+def main():
+    port = int(sys.argv[1])
+    srv = http.server.ThreadingHTTPServer(("127.0.0.1", port), Handler)
+    print(f"Fixture serving on port {port}", flush=True)
+    srv.serve_forever()
+
+
+if __name__ == "__main__":
+    main()

.gitea/scripts/tests/test_review_refire_status.sh

@@ -0,0 +1,152 @@
+#!/usr/bin/env bash
+# Regression tests for .gitea/scripts/review-refire-status.sh.
+#
+# review-refire-status.sh fetches the PR head SHA, runs review-check.sh,
+# and POSTs a status context based on review-check.sh's exit code.
+#
+# Test matrix:
+#   T3  — closed PR no-op  (requires jq + HTTP fixture)
+#   T4  — GET /pulls/{N} non-200 → exits 1 (connection refused)
+#   T6  — missing required env → exits 1
+#   T7  — bash syntax check (bash -n passes)
+#   T8  — auth file security (mode 600 + Authorization header)
+#
+# T1/T2 (review-check success/failure) require jq and are run via
+# the GitHub Actions CI job (jq installed via apt-get).
+#
+# Hostile self-review: MUST FAIL if script is absent.
+
+set -euo pipefail
+
+THIS_DIR="$(cd "$(dirname "$0")" && pwd)"
+SCRIPT_DIR="$(cd "$THIS_DIR/.." && pwd)"
+SCRIPT="$SCRIPT_DIR/review-refire-status.sh"
+
+PASS=0
+FAIL=0
+FAILED_TESTS=""
+
+pass() { echo "  PASS: $1"; PASS=$((PASS+1)); }
+fail() { echo "  FAIL: $1"; FAIL=$((FAIL+1)); FAILED_TESTS="${FAILED_TESTS} $1"; }
+
+if [ ! -f "$SCRIPT" ]; then
+    echo "FAIL: $SCRIPT does not exist"
+    exit 1
+fi
+
+# T7: bash syntax
+t7_syntax() {
+    if bash -n "$SCRIPT" 2>&1; then
+        pass "T7 bash syntax"
+    else
+        fail "T7 bash syntax"
+    fi
+}
+
+# T6: missing required env
+t6_missing_env() {
+    set +e
+    local out rc
+    out=$(GITEA_TOKEN="x" GITEA_HOST="git.example" REPO="o/r" PR_NUMBER="1" \
+          TEAM="qa" COMMENT_AUTHOR="alice" bash "$SCRIPT" 2>&1)
+    rc=$?
+    set -e
+    if [ "$rc" -ne 0 ]; then
+        pass "T6 missing GITEA_TOKEN exits non-zero (rc=$rc)"
+    else
+        fail "T6 missing GITEA_TOKEN should exit non-zero"
+    fi
+}
+
+# T4: connection refused (port nothing listening on)
+t4_connection_refused() {
+    local port
+    port=$(python3 -c "import socket; s=socket.socket(); s.bind(('127.0.0.1',0)); print(s.getsockname()[1]); s.close()")
+    # Don't start a server on this port
+
+    set +e
+    local out rc
+    out=$(GITEA_TOKEN="test-token" \
+          GITEA_HOST="127.0.0.1:${port}" \
+          REPO="molecule-ai/molecule-core" \
+          PR_NUMBER="1" \
+          TEAM="qa" \
+          COMMENT_AUTHOR="alice" \
+          bash "$SCRIPT" 2>&1)
+    rc=$?
+    set -e
+
+    if [ "$rc" -ne 0 ]; then
+        pass "T4 connection refused exits non-zero (rc=$rc)"
+    else
+        fail "T4 connection refused should exit non-zero, got rc=0"
+    fi
+}
+
+# T8: auth file security
+t8_auth_security() {
+    if grep -q 'chmod 600' "$SCRIPT"; then
+        pass "T8 auth file mode 600"
+    else
+        fail "T8 should chmod 600 on auth file"
+    fi
+    if grep -q 'Authorization.*token' "$SCRIPT"; then
+        pass "T8 Authorization header set"
+    else
+        fail "T8 should set Authorization header"
+    fi
+}
+
+# T3: closed PR no-op — requires jq (installed in CI via apt-get)
+t3_closed_pr_noop() {
+    if ! command -v jq >/dev/null 2>&1; then
+        echo "  SKIP T3: jq not available (run in CI to exercise)"
+        return
+    fi
+    local port fixture_dir out rc
+    port=$(python3 -c "import socket; s=socket.socket(); s.bind(('127.0.0.1',0)); print(s.getsockname()[1]); s.close()")
+    fixture_dir=$(mktemp -d)
+    echo "closed" > "${fixture_dir}/scenario"
+    export FIXTURE_STATE_DIR="$fixture_dir"
+
+    python3 "$THIS_DIR/_review_refire_fixture.py" "$port" &
+    local fixture_pid=$!
+    sleep 1
+
+    out=$(GITEA_TOKEN="test-token" \
+          GITEA_HOST="127.0.0.1:${port}" \
+          REPO="molecule-ai/molecule-core" \
+          PR_NUMBER="1" \
+          TEAM="qa" \
+          COMMENT_AUTHOR="alice" \
+          bash "$SCRIPT" 2>&1)
+    rc=$?
+
+    kill $fixture_pid 2>/dev/null || true
+    rm -rf "$fixture_dir"
+
+    if [ "$rc" -eq 0 ]; then
+        pass "T3 closed PR exits 0"
+    else
+        fail "T3 closed PR should exit 0, got rc=$rc. Output: ${out}"
+    fi
+}
+
+echo "Running review-refire-status.sh tests..."
+echo "========================================"
+
+t7_syntax
+t6_missing_env
+t4_connection_refused
+t8_auth_security
+t3_closed_pr_noop
+
+echo ""
+echo "========================================"
+echo "PASS: $PASS  FAIL: $FAIL"
+if [ "$FAIL" -gt 0 ]; then
+    echo "Failed:$FAILED_TESTS"
+    exit 1
+fi
+echo "All tests passed."
+exit 0

.gitea/scripts/tests/test_sop_checklist.py

@@ -575,22 +575,29 @@ class TestComputeNaState(unittest.TestCase):
         comments = [_comment("bob", "/sop-n/a qa-review N/A: pure tooling change")]
         na_state = sop.compute_na_state(comments, "alice", na_gates, lambda g, u: u)
         self.assertTrue(na_state["qa-review"]["declared"])
-        self.assertEqual(na_state["qa-review"]["decl_ackers"], ["bob"])
+        self.assertEqual(na_state["qa-review"]["declared_by"], "bob")
+        self.assertTrue(na_state["qa-review"]["valid"])
+        self.assertIsNone(na_state["qa-review"]["error"])
 
     def test_na_declared_by_unauthorized_user_rejected(self):
         cfg = sop.load_config(CONFIG_PATH)
         na_gates = cfg.get("n/a_gates", {})
         comments = [_comment("mallory", "/sop-n/a qa-review N/A: not real team")]
         na_state = sop.compute_na_state(comments, "alice", na_gates, lambda g, u: [])
-        self.assertFalse(na_state["qa-review"]["declared"])
-        self.assertEqual(na_state["qa-review"]["rejected"]["not_in_team"], ["mallory"])
+        # declared=True (mallory attempted N/A) but valid=False (not in required team).
+        self.assertTrue(na_state["qa-review"]["declared"])
+        self.assertFalse(na_state["qa-review"]["valid"])
+        self.assertIn("mallory", na_state["qa-review"]["error"])
 
     def test_author_cannot_self_declare_na(self):
         cfg = sop.load_config(CONFIG_PATH)
         na_gates = cfg.get("n/a_gates", {})
         comments = [_comment("alice", "/sop-n/a qa-review N/A: I am the author")]
         na_state = sop.compute_na_state(comments, "alice", na_gates, lambda g, u: u)
-        self.assertFalse(na_state["qa-review"]["declared"])
+        # declared=True (alice attempted N/A) but valid=False (self-declare rejected).
+        self.assertTrue(na_state["qa-review"]["declared"])
+        self.assertFalse(na_state["qa-review"]["valid"])
+        self.assertIn("self", na_state["qa-review"]["error"].lower())
 
     def test_parse_directives_separates_na_from_ack(self):
         directives, na_directives = sop.parse_directives(

.gitea/workflows/review-refire-comments.yml

@@ -70,7 +70,10 @@ jobs:
       - name: Refire qa-review status
         if: steps.classify.outputs.run_qa == 'true'
         env:
-          GITEA_TOKEN: ${{ secrets.RFC_324_TEAM_READ_TOKEN || secrets.GITHUB_TOKEN }}
+          # RFC_324_TEAM_READ_TOKEN is read-only (team membership read scope only).
+          # review-refire-status.sh POSTs to /statuses — requires write scope.
+          # SOP_TIER_CHECK_TOKEN carries write:repository + write:issue + read:organization.
+          GITEA_TOKEN: ${{ secrets.SOP_TIER_CHECK_TOKEN || secrets.GITHUB_TOKEN }}
           GITEA_HOST: git.moleculesai.app
           REPO: ${{ github.repository }}
           PR_NUMBER: ${{ github.event.issue.number }}
@@ -87,7 +90,10 @@ jobs:
       - name: Refire security-review status
         if: steps.classify.outputs.run_security == 'true'
         env:
-          GITEA_TOKEN: ${{ secrets.RFC_324_TEAM_READ_TOKEN || secrets.GITHUB_TOKEN }}
+          # RFC_324_TEAM_READ_TOKEN is read-only (team membership read scope only).
+          # review-refire-status.sh POSTs to /statuses — requires write scope.
+          # SOP_TIER_CHECK_TOKEN carries write:repository + write:issue + read:organization.
+          GITEA_TOKEN: ${{ secrets.SOP_TIER_CHECK_TOKEN || secrets.GITHUB_TOKEN }}
           GITEA_HOST: git.moleculesai.app
           REPO: ${{ github.repository }}
           PR_NUMBER: ${{ github.event.issue.number }}

.gitea/workflows/review-refire-status-tests.yml

@@ -0,0 +1,63 @@
+name: review-refire-status-tests
+
+# Regression tests for .gitea/scripts/review-refire-status.sh.
+#
+# review-refire-status.sh is load-bearing: it POSTs the qa-review and
+# security-review slash-command status contexts to the PR head SHA.
+# It calls review-check.sh, which requires jq.
+#
+# Design (mirrors review-check-tests.yml):
+#   - Bash test harness; custom assert framework (no bats dependency).
+#   - jq is required (used by review-check.sh which this script invokes).
+#   - continue-on-error: false — these tests must pass.
+
+on:
+  push:
+    branches: [main, staging]
+    paths:
+      - '.gitea/scripts/review-refire-status.sh'
+      - '.gitea/scripts/tests/test_review_refire_status.sh'
+      - '.gitea/scripts/tests/_review_refire_fixture.py'
+      - '.gitea/workflows/review-refire-status-tests.yml'
+  pull_request:
+    branches: [main, staging]
+    paths:
+      - '.gitea/scripts/review-refire-status.sh'
+      - '.gitea/scripts/tests/test_review_refire_status.sh'
+      - '.gitea/scripts/tests/_review_refire_fixture.py'
+      - '.gitea/workflows/review-refire-status-tests.yml'
+  workflow_dispatch:
+
+env:
+  GITHUB_SERVER_URL: https://git.moleculesai.app
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  # bp-exempt: regression tests for review-refire-status.sh; informational only, not a merge gate.
+  test:
+    name: review-refire-status.sh regression tests
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@de0fac2e4500dabe0009e67214ff5f5447ce83dd # v6.0.2
+
+      - name: Install jq
+        # review-refire-status.sh invokes review-check.sh, which uses jq for
+        # JSON parsing. Gitea Actions runners do not bundle jq.
+        # mc#774: pre-existing continue-on-error mask; root-fix and remove, do not renew silently.
+        continue-on-error: true
+        run: |
+          if apt-get update -qq && apt-get install -y -qq jq; then
+            echo "::notice::jq installed via apt-get: $(jq --version)"
+          elif timeout 120 curl -sSL \
+            "https://github.com/jqlang/jq/releases/download/jq-1.7.1/jq-linux-amd64" \
+            -o /usr/local/bin/jq && chmod +x /usr/local/bin/jq; then
+            echo "::notice::jq binary downloaded: $(/usr/local/bin/jq --version)"
+          else
+            echo "::warning::jq install failed"
+          fi
+
+      - name: Run review-refire-status.sh regression suite
+        run: bash .gitea/scripts/tests/test_review_refire_status.sh

tests/test_main_red_watchdog.py

@@ -410,27 +410,68 @@ def test_auto_close_when_main_returns_to_green(wd_module, monkeypatch):
     assert patch_calls[0][2] == {"state": "closed"}
 
 
-def test_auto_close_skips_when_main_pending(wd_module, monkeypatch):
-    """main pending (CI still running) at NEW_SHA → leave old issue alone.
-    Pending could resolve to red, so closing prematurely would lose the
-    breadcrumb of the prior red."""
-    old_title = f"[main-red] owner/repo: {SHA_RED[:10]}"
+def test_auto_close_on_main_pending_with_no_failures(wd_module, monkeypatch):
+    """main pending at NEW_SHA but all individual statuses are successful
+    (some jobs still running, others done) → is_red() confirms 0 failures,
+    so close stale issues. The combined `pending` state alone is insufficient
+    to block closing when individual entries show no failures."""
     stub = _make_stub_api({
         ("GET", "/repos/owner/repo/branches/main"): (200, _branches_response(SHA_GREEN)),
         ("GET", f"/repos/owner/repo/commits/{SHA_GREEN}/status"): (
             200, _combined_status("pending", [
+                # Some still running, some already succeeded
                 {"context": "ci/test", "state": "pending"},
+                {"context": "ci/lint", "state": "success"},
             ]),
         ),
+        # Stale issue from old SHA
+        ("GET", "/repos/owner/repo/issues"): (
+            200, [{"number": 7, "title": f"[main-red] owner/repo: {SHA_RED[:10]}"}],
+        ),
+        ("POST", "/repos/owner/repo/issues/7/comments"): (200, {}),
+        ("PATCH", "/repos/owner/repo/issues/7"): (200, {}),
     })
     monkeypatch.setattr(wd_module, "api", stub)
 
     wd_module.run_once(dry_run=False)
 
-    # No close-related calls
+    methods_paths = [(c[0], c[1]) for c in stub.calls]
+    assert ("GET", "/repos/owner/repo/issues") in methods_paths
+    assert ("POST", "/repos/owner/repo/issues/7/comments") in methods_paths
+    assert ("PATCH", "/repos/owner/repo/issues/7") in methods_paths
+
+
+def test_auto_close_skips_when_main_pending_with_failures(wd_module, monkeypatch):
+    """main pending at NEW_SHA but some individual statuses are failures
+    (CI partially broken, combined state lagging) → is_red() sees failures,
+    so close must NOT run. file_or_update_red handles the new failure
+    (opens an issue for the current SHA)."""
+    stub = _make_stub_api({
+        ("GET", "/repos/owner/repo/branches/main"): (200, _branches_response(SHA_GREEN)),
+        ("GET", f"/repos/owner/repo/commits/{SHA_GREEN}/status"): (
+            200, _combined_status("pending", [
+                {"context": "ci/test", "state": "pending"},
+                {"context": "ci/lint", "state": "failure"},  # individual failure
+            ]),
+        ),
+        # file_or_update_red: search for existing issue for current SHA → none found
+        ("GET", "/repos/owner/repo/issues"): (200, []),
+        # file_or_update_red: POST new issue
+        ("POST", "/repos/owner/repo/issues"): (200, {"number": 8}),
+        # file_or_update_red: GET labels
+        ("GET", "/repos/owner/repo/labels"): (200, [{"id": 9, "name": "tier:high"}]),
+        # file_or_update_red: POST label
+        ("POST", "/repos/owner/repo/issues/8/labels"): (200, {}),
+    })
+    monkeypatch.setattr(wd_module, "api", stub)
+
+    wd_module.run_once(dry_run=False)
+
+    # No close-related calls — is_red() triggered file/update path instead
     methods_paths = [(c[0], c[1]) for c in stub.calls]
     assert ("PATCH", "/repos/owner/repo/issues/7") not in methods_paths
-    assert ("GET", "/repos/owner/repo/issues") not in methods_paths
+    assert ("POST", "/repos/owner/repo/issues/7/comments") not in methods_paths
+    assert ("GET", "/repos/owner/repo/issues/7") not in methods_paths
 
 
 # --------------------------------------------------------------------------

tests/test_sop_checklist.py

@@ -0,0 +1,399 @@
+"""Tests for `.gitea/scripts/sop-checklist.py`.
+
+Covers the N/A declarations feature (`compute_na_state`) and the
+`parse_directives` helper that powers both ack/revoke and N/A parsing.
+
+Run:
+    python3 -m pytest tests/test_sop_checklist.py -v
+
+Dependencies: stdlib + pytest. No network. No live Gitea calls.
+"""
+from __future__ import annotations
+
+import importlib.util
+import os
+import sys
+from pathlib import Path
+from unittest import mock
+
+import pytest
+
+
+# --------------------------------------------------------------------------
+# Module-import fixture
+# --------------------------------------------------------------------------
+SCRIPT_PATH = (
+    Path(__file__).resolve().parent.parent
+    / ".gitea"
+    / "scripts"
+    / "sop-checklist.py"
+)
+
+
+@pytest.fixture(scope="module")
+def sc_module():
+    """Import sop-checklist.py as a module under a known env."""
+    env = {
+        "GITEA_TOKEN": "test-token",
+        "GITEA_HOST": "git.example.test",
+        "REPO": "owner/repo",
+        "DRY_RUN": "0",
+    }
+    with mock.patch.dict(os.environ, env, clear=False):
+        spec = importlib.util.spec_from_file_location(
+            "sop_checklist", SCRIPT_PATH
+        )
+        m = importlib.util.module_from_spec(spec)
+        spec.loader.exec_module(m)
+        # Force-set module globals captured at import time.
+        m.GITEA_TOKEN = env["GITEA_TOKEN"]
+        m.GITEA_HOST = env["GITEA_HOST"]
+        m.REPO = env["REPO"]
+        yield m
+
+
+# --------------------------------------------------------------------------
+# parse_directives
+# --------------------------------------------------------------------------
+
+class TestParseDirectives:
+    def test_empty_body(self, sc_module):
+        directives, na_directives = sc_module.parse_directives("", {})
+        assert directives == []
+        assert na_directives == []
+
+    def test_ack_directive(self, sc_module):
+        directives, na_directives = sc_module.parse_directives(
+            "/sop-ack qa-review", {}
+        )
+        assert directives == [("sop-ack", "qa-review", "")]
+        assert na_directives == []
+
+    def test_ack_directive_with_note(self, sc_module):
+        directives, na_directives = sc_module.parse_directives(
+            "/sop-ack qa-review CI-only change", {}
+        )
+        assert directives == [("sop-ack", "qa-review", "CI-only change")]
+        assert na_directives == []
+
+    def test_revoke_directive(self, sc_module):
+        directives, na_directives = sc_module.parse_directives(
+            "/sop-revoke qa-review missed edge case", {}
+        )
+        assert directives == [("sop-revoke", "qa-review", "missed edge case")]
+        assert na_directives == []
+
+    def test_na_directive_no_reason(self, sc_module):
+        """Bare /sop-n/a <gate> is valid."""
+        _, na_directives = sc_module.parse_directives(
+            "/sop-n/a qa-review", {}
+        )
+        assert na_directives == [("qa-review", "")]
+
+    def test_na_directive_with_reason(self, sc_module):
+        """Full /sop-n/a <gate> <reason> is parsed correctly."""
+        _, na_directives = sc_module.parse_directives(
+            "/sop-n/a qa-review CI/non-security-touching", {}
+        )
+        assert na_directives == [("qa-review", "CI/non-security-touching")]
+
+    def test_na_directive_multiple_gates(self, sc_module):
+        body = (
+            "/sop-n/a qa-review CI-only\n"
+            "/sop-n/a security-review no-op change\n"
+        )
+        _, na_directives = sc_module.parse_directives(body, {})
+        assert na_directives == [
+            ("qa-review", "CI-only"),
+            ("security-review", "no-op change"),
+        ]
+
+    def test_na_directive_requires_space_after_gate(self, sc_module):
+        """Without a space, /sop-n/a is not matched (the regex requires \t+)."""
+        _, na_directives = sc_module.parse_directives(
+            "/sop-n/a", {}
+        )
+        assert na_directives == []
+
+    def test_na_directive_invalid_gate_characters(self, sc_module):
+        r"""Gate names must match [A-Za-z0-9_\-]+. A dot in the gate name
+        causes the entire line not to match (no backtracking to partial match)."""
+        _, na_directives = sc_module.parse_directives(
+            "/sop-n/a qa.review some reason", {}
+        )
+        # The regex cannot match qa.review as a valid gate (dot is invalid).
+        # Since the (?:[ \t]+(.*))? is optional, the whole pattern fails and
+        # no directive is extracted.
+        assert na_directives == []
+
+    def test_mixed_directives(self, sc_module):
+        """ack + na directives can appear in the same comment."""
+        body = (
+            "/sop-ack sop-checklist all boxes ticked\n"
+            "/sop-n/a qa-review no code changes\n"
+        )
+        directives, na_directives = sc_module.parse_directives(body, {})
+        assert directives == [("sop-ack", "sop-checklist", "all boxes ticked")]
+        assert na_directives == [("qa-review", "no code changes")]
+
+    def test_na_directive_leading_whitespace(self, sc_module):
+        """Indent is permitted — regex matches [ \t]*."""
+        _, na_directives = sc_module.parse_directives(
+            "  /sop-n/a qa-review indented", {}
+        )
+        assert na_directives == [("qa-review", "indented")]
+
+    def test_na_directive_tab_indent(self, sc_module):
+        _, na_directives = sc_module.parse_directives(
+            "\t/sop-n/a qa-review tab-indented", {}
+        )
+        assert na_directives == [("qa-review", "tab-indented")]
+
+    def test_na_directive_trailing_whitespace_stripped(self, sc_module):
+        """Trailing spaces after the reason are stripped."""
+        _, na_directives = sc_module.parse_directives(
+            "/sop-n/a qa-review reason   \n", {}
+        )
+        assert na_directives == [("qa-review", "reason")]
+
+    def test_na_directive_multiple_on_same_line(self, sc_module):
+        """MULTILINE anchors at line start (^), but greedy (.*) in the reason
+        group captures everything after the first space, including the second
+        /sop-n/a on the same line."""
+        _, na_directives = sc_module.parse_directives(
+            "/sop-n/a qa-review /sop-n/a security-review", {}
+        )
+        # reason = "/sop-n/a security-review" (greedy capture)
+        assert na_directives == [("qa-review", "/sop-n/a security-review")]
+
+    def test_ack_numeric_alias(self, sc_module):
+        """Numeric alias e.g. /sop-ack 1 resolves via numeric_aliases."""
+        directives, _ = sc_module.parse_directives(
+            "/sop-ack 1", {1: "qa-review"}
+        )
+        assert directives == [("sop-ack", "qa-review", "")]
+
+
+# --------------------------------------------------------------------------
+# compute_na_state
+# --------------------------------------------------------------------------
+
+class TestComputeNaState:
+    def _make_comment(
+        self,
+        body: str,
+        user_login: str,
+        created_at: str = "2024-01-01T00:00:00Z",
+    ) -> dict:
+        return {
+            "id": 0,
+            "body": body,
+            "user": {"login": user_login},
+            "created_at": created_at,
+        }
+
+    def _na_gates(self) -> dict:
+        return {
+            "qa-review": {
+                "required_teams": ["qa"],
+            },
+            "security-review": {
+                "required_teams": ["security"],
+            },
+            "sop-checklist": {
+                "required_teams": ["engineering"],
+            },
+        }
+
+    # ----- empty / no declarations -----
+
+    def test_no_comments(self, sc_module):
+        result = sc_module.compute_na_state([], "author", self._na_gates(), lambda *a: [])
+        for gate in ["qa-review", "security-review"]:
+            assert result[gate]["declared"] is False
+            assert result[gate]["valid"] is False
+            assert result[gate]["error"] is None
+
+    def test_non_na_comment_ignored(self, sc_module):
+        comments = [self._make_comment("/sop-ack qa-review", "helper")]
+        result = sc_module.compute_na_state(comments, "author", self._na_gates(), lambda *a: [])
+        assert result["qa-review"]["declared"] is False
+
+    def test_unknown_gate_ignored(self, sc_module):
+        comments = [self._make_comment("/sop-n/a unknown-gate reason", "helper")]
+        result = sc_module.compute_na_state(comments, "author", self._na_gates(), lambda *a: [])
+        assert result["qa-review"]["declared"] is False
+
+    # ----- valid declarations -----
+
+    def test_valid_na_declaration(self, sc_module):
+        """Declarer in required team → valid=True, declared=True."""
+        comments = [self._make_comment("/sop-n/a qa-review CI-only change", "qa-member")]
+        probe = lambda _slug, users: ["qa-member"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is True
+        assert result["qa-review"]["declared_by"] == "qa-member"
+        assert result["qa-review"]["reason"] == "CI-only change"
+        assert result["qa-review"]["valid"] is True
+        assert result["qa-review"]["error"] is None
+
+    def test_valid_na_multiple_gates(self, sc_module):
+        comments = [
+            self._make_comment("/sop-n/a qa-review CI-only", "qa-member"),
+            self._make_comment("/sop-n/a security-review no-op", "sec-member"),
+        ]
+        probe = lambda slug, users: (
+            ["qa-member"] if "qa" in slug else ["sec-member"]
+        )
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["valid"] is True
+        assert result["security-review"]["valid"] is True
+
+    def test_na_without_reason(self, sc_module):
+        """N/A with no reason is still valid if team membership checks out."""
+        comments = [self._make_comment("/sop-n/a qa-review", "qa-member")]
+        probe = lambda _slug, users: ["qa-member"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is True
+        assert result["qa-review"]["reason"] == ""
+        assert result["qa-review"]["valid"] is True
+
+    # ----- invalid declarations -----
+
+    def test_self_declare_rejected(self, sc_module):
+        """Authors cannot self-declare N/A."""
+        comments = [self._make_comment("/sop-n/a qa-review self-declared", "author")]
+        probe = lambda _slug, users: ["author"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is True
+        assert result["qa-review"]["valid"] is False
+        assert result["qa-review"]["error"] == "self-declare N/A rejected"
+
+    def test_not_in_required_team(self, sc_module):
+        """Declarer not in required team → valid=False, error set."""
+        comments = [self._make_comment("/sop-n/a qa-review CI-only", "random-user")]
+        probe = lambda _slug, users: []  # nobody approved
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is True
+        assert result["qa-review"]["valid"] is False
+        assert "not in required team" in result["qa-review"]["error"]
+
+    def test_probe_returns_empty_not_member(self, sc_module):
+        """Probe returns empty list even though the team exists."""
+        comments = [self._make_comment("/sop-n/a security-review no-op", "not-sec")]
+        probe = lambda _slug, users: []
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["security-review"]["valid"] is False
+        assert "not in required team" in result["security-review"]["error"]
+
+    # ----- chronological ordering -----
+
+    def test_most_recent_na_wins_per_user(self, sc_module):
+        """For the same user+gate, the most-recent comment wins."""
+        comments = [
+            self._make_comment("/sop-n/a qa-review old reason", "qa-member",
+                              created_at="2024-01-01T00:00:00Z"),
+            self._make_comment("/sop-n/a qa-review newer reason", "qa-member",
+                              created_at="2024-01-02T00:00:00Z"),
+        ]
+        probe = lambda _slug, users: ["qa-member"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["reason"] == "newer reason"
+
+    def test_different_users_same_gate_both_declared(self, sc_module):
+        """Two different declarers for the same gate — both recorded."""
+        comments = [
+            self._make_comment("/sop-n/a qa-review first declarer", "qa-member-1",
+                              created_at="2024-01-01T00:00:00Z"),
+            self._make_comment("/sop-n/a qa-review second declarer", "qa-member-2",
+                              created_at="2024-01-02T00:00:00Z"),
+        ]
+        probe = lambda _slug, users: ["qa-member-1", "qa-member-2"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        # Most-recent declaration wins (qa-member-2 is newer).
+        assert result["qa-review"]["declared_by"] == "qa-member-2"
+        assert result["qa-review"]["reason"] == "second declarer"
+        assert result["qa-review"]["valid"] is True
+
+    # ----- gate key is exact -----
+
+    def test_unknown_gate_in_na_directive_ignored(self, sc_module):
+        """A gate name not in na_gates config is silently ignored."""
+        comments = [
+            self._make_comment("/sop-n/a qa-review valid", "qa-member"),
+            self._make_comment("/sop-n/a totally-unknown-gate reason", "qa-member"),
+        ]
+        probe = lambda _slug, users: ["qa-member"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is True
+        # unknown-gate is not in result keys — only configured gates appear.
+        assert "totally-unknown-gate" not in result
+
+    # ----- empty comments list -----
+
+    def test_comment_with_no_user_field(self, sc_module):
+        """A comment dict with no user.login is skipped (no crash)."""
+        comments = [{"id": 1, "body": "/sop-n/a qa-review", "user": {}}]
+        probe = lambda _slug, users: ["qa-member"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is False
+
+    def test_comment_with_empty_body(self, sc_module):
+        """A comment with empty body is handled gracefully."""
+        comments = [{"id": 1, "body": "", "user": {"login": "qa-member"}}]
+        probe = lambda _slug, users: ["qa-member"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is False
+
+    # ----- probe slug format -----
+
+    def test_probe_called_with_na_prefix(self, sc_module):
+        """The team membership probe is called with slug='na:<gate>'."""
+        calls = []
+        def track_probe(slug, users):
+            calls.append((slug, users))
+            return []
+
+        comments = [self._make_comment("/sop-n/a qa-review reason", "qa-member")]
+        sc_module.compute_na_state(
+            comments, "author", self._na_gates(), track_probe
+        )
+        assert ("na:qa-review", ["qa-member"]) in calls
+
+    # ----- undeclared gate remains False -----
+
+    def test_undeclared_gate_preserved(self, sc_module):
+        """A gate with no declaration stays declared=False even when
+        other gates are declared."""
+        comments = [
+            self._make_comment("/sop-n/a qa-review CI-only", "qa-member"),
+        ]
+        probe = lambda _slug, users: ["qa-member"]
+        result = sc_module.compute_na_state(
+            comments, "author", self._na_gates(), probe
+        )
+        assert result["qa-review"]["declared"] is True
+        assert result["security-review"]["declared"] is False
+        assert result["sop-checklist"]["declared"] is False