fix(hermes): guard the hand-over claim window; skip repos whose permission lookup fails

review-bot on b6179a41:
- Expert card bodies open with "This card is for the <expert> expert. If you are
  not <expert>, do no work on it": a coordinator that claims the card between the
  body edit and the reassign completes it instead of running the alert protocol
  next to the separate @expert card.
- issue_repos skips (and names) a repo whose hermes-bot permission lookup fails
  instead of aborting; the other repos still get labels and backfill, the Job
  still fails.

kubernetes/apps/hermes/skills/gitea-issues/scripts/gitea_issues.py

@@ -211,12 +211,16 @@ def card_from_payload(payload: dict) -> dict:
         ])
 
     routing_body = render(routing_section(full_name, number))
+    # Guards the edit-then-assign window of hand_over: a coordinator that claims the card
+    # with this body already on it must not do the expert's work.
+    guard = [f"This card is for the {assignee} expert. If you are not {assignee}, do no work on it",
+             f"and complete it with one line: \"Handed over to {assignee}\"."]
     if assignee == COORDINATOR:
         body = routing_body
     elif title.startswith(ALERT_PREFIX):
-        body = render(alert_section(full_name, number))
+        body = render([*guard, "", *alert_section(full_name, number)])
     else:
-        body = render([])
+        body = render(guard)
     return {
         "owner": owner, "repo": name, "number": number,
         "key": f"{full_name}#{number}",

kubernetes/apps/hermes/skills/gitea-issues/scripts/sync_webhook.py

@@ -138,15 +138,22 @@ def reconcile(token: str, secret: str) -> str:
 
 # --- 2. domain labels --------------------------------------------------------
 
-def issue_repos(token: str) -> list[dict]:
+def issue_repos(token: str, errors: list[str] | None = None) -> list[dict]:
     """Repos whose issues Hermes works: issue tracker on, not archived, and hermes-bot may
-    write issues there (it sets the labels and comments). A site admin sees every repo."""
+    write issues there (it sets the labels and comments). A site admin sees every repo.
+    A repo whose permission lookup fails is skipped and named in `errors`."""
     out = []
     for r in paged("/repos/search", token):
         if not r.get("has_issues", True) or r.get("archived"):
             continue
         owner, _, name = r["full_name"].partition("/")
-        perm = api("GET", f"/repos/{q(owner)}/{q(name)}/collaborators/{q(BOT_LOGIN)}/permission", token)
+        try:
+            perm = api("GET", f"/repos/{q(owner)}/{q(name)}/collaborators/{q(BOT_LOGIN)}/permission", token)
+        except RuntimeError as exc:
+            if errors is None:
+                raise
+            errors.append(f"{r['full_name']}: {exc}")
+            continue
         if (perm or {}).get("permission") in BOT_WRITE:
             out.append(r)
     return out
@@ -252,8 +259,12 @@ def main() -> int:
     repos: list[dict] = []
 
     def load_repos() -> str:
-        repos.extend(issue_repos(token))
-        return f"{len(repos)} with issues writable by {BOT_LOGIN}"
+        errors: list[str] = []
+        repos.extend(issue_repos(token, errors))
+        msg = f"{len(repos)} with issues writable by {BOT_LOGIN}"
+        if errors:  # the other repos still get labels and backfill; the Job fails
+            raise RuntimeError(f"{msg}; permission lookup failed, skipped: {'; '.join(errors)}")
+        return msg
 
     failures = 0
     for name, step in (("webhook", lambda: reconcile(token, secret)),

kubernetes/apps/hermes/skills/tests/test_gitea_issues.py

@@ -40,6 +40,8 @@ class Routing(unittest.TestCase):
         card = gi.card_from_payload(payload(labels=["bug", "Atlas"]))
         self.assertEqual(card["assignee"], "atlas")
         self.assertNotIn("Routing (you are the coordinator)", card["body"])
+        self.assertIn("This card is for the atlas expert.", card["body"])
+        self.assertNotIn("This card is for the", card["routing_body"])
 
     def test_no_domain_label_is_coordinator_with_routing_instructions(self):
         card = gi.card_from_payload(payload(labels=["bug"]))
@@ -190,6 +192,7 @@ class HandOver(unittest.TestCase):
         self.assertEqual(calls[1][1], "t_c")
         self.assertNotIn("Routing (you are the coordinator)", body)
         self.assertIn("alert-fix/issue-7", body)
+        self.assertIn("If you are not atlas, do no work on it", body)  # guards the claim window
         self.assertIn("hermes-kanban-card:t_c:atlas", posts[0][2]["body"])
 
     def test_failed_edit_never_hands_routing_text_to_the_expert(self):
@@ -351,6 +354,21 @@ class DomainLabels(unittest.TestCase):
         self.assertIn("/repos/a/x/collaborators/hermes-bot/permission", seen)
         self.assertFalse([p for p in seen if p.startswith(("/repos/a/y/", "/repos/a/z/"))])
 
+    def test_failed_permission_lookup_skips_only_that_repo(self):
+        def fake(method, path, token, body=None):
+            if path.startswith("/repos/search"):
+                return {"data": [repo("a/x"), repo("a/broken"), repo("a/w")]}
+            if path.startswith("/repos/a/broken/"):
+                raise RuntimeError("GET /repos/a/broken/...: HTTP 500")
+            return {"permission": "write"}
+
+        errors = []
+        with mock.patch.object(sync, "api", side_effect=fake):
+            got = [r["full_name"] for r in sync.issue_repos("tok", errors)]
+        self.assertEqual(got, ["a/x", "a/w"])
+        self.assertEqual(len(errors), 1)
+        self.assertIn("a/broken", errors[0])
+
 
 class Backfill(unittest.TestCase):
     def run_backfill(self, issues, carded=(), status=200, limit=10, budget=900.0):