kubernetes/apps/hermes/skills/gitea-issues/SKILL.md

@@ -17,8 +17,11 @@ Every open issue on gitea.renner.uno becomes one kanban card, without a manual s
 1. A Gitea system webhook (name `hermes-kanban-<fingerprint>`, events: issues and
    issue labels) posts to `http://hermes.hermes.svc.cluster.local:8644/webhooks/gitea-issues`.
    The CronJob `gitea-webhook-sync` (namespace hermes, every 15 min) keeps that hook in
-   place, creates the domain labels and backfills, both only in repos where hermes-bot
-   has issue write (org labels for their orgs, repo labels for user-owned ones): open
+   place, creates the domain labels and backfills, both in every non-archived repo with
+   issues of the orgs hermes-bot belongs to, where an org team grants it issue write
+   (today: all repos of `konairius` through team `Bots`). Labels are org labels only.
+   User-owned repos and repos where hermes-bot is only a direct collaborator get neither
+   labels nor backfill; their new issues still become cards through the system hook. Backfill: open
    issues without a card comment are replayed to the route as signed `opened`
    deliveries, at most 10 per run. Its site-admin PAT lives at OpenBao
    `apps/gitea-webhook-sync/admin`, outside what the agent's role can read.

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

@@ -7,10 +7,10 @@
    active system hook carries the desired name, a new one is created first and every
    other hook with the prefix is deleted afterwards.
 2. Domain labels. hestia/atlas/apollo/clio/daedalus are created only where Hermes
-   works issues: repos with an issue tracker in which BOT_LOGIN (hermes-bot) has
-   issue write. They become org labels in each org owning such a repo and repo labels
-   in such user-owned repos (org labels do not reach those). Existing labels are left
-   as they are; only missing ones are created.
+   works issues: the non-archived org repos with an issue tracker in which BOT_LOGIN
+   (hermes-bot) may write issues through an org team (it labels and comments). They
+   are org labels of those orgs, never repo labels; user-owned repos are out of scope.
+   Existing labels are left as they are; only missing ones are created.
 3. Backfill. Open issues of the same repos without the bridge's card comment are
    replayed to the Hermes webhook route as signed `opened` deliveries, at most
    BACKFILL_LIMIT per run and none started after RUN_BUDGET seconds, in random order
@@ -49,9 +49,10 @@ BACKFILL_SPACING = float(os.environ.get("BACKFILL_SPACING_SECONDS", "3"))
 RUN_BUDGET = float(os.environ.get("RUN_BUDGET_SECONDS", "900"))
 DELIVER_TIMEOUT = 60
 CARD_MARKER = "<!-- hermes-kanban-card:"
-# Labels and backfill only where this account may write issues (it labels and comments).
+# Labels and backfill only in org repos this account may write issues in (it labels and comments).
 BOT_LOGIN = os.environ.get("BOT_LOGIN", "hermes-bot")
 BOT_WRITE = ("write", "admin", "owner")
+ISSUES_UNIT = "repo.issues"
 STARTED = time.monotonic()
 
 DOMAIN_LABELS = {
@@ -88,13 +89,11 @@ def api(method: str, path: str, token: str, body: dict | None = None):
 
 
 def paged(path: str, token: str) -> list[dict]:
-    """All pages of a list endpoint; /repos/search wraps its list in {"data": [...]}."""
+    """All pages of a list endpoint."""
     out, page = [], 1
     sep = "&" if "?" in path else "?"
     while True:
         batch = api("GET", f"{path}{sep}page={page}&limit=50", token) or []
-        if isinstance(batch, dict):
-            batch = batch.get("data") or []
         out.extend(batch)
         if len(batch) < 50:
             return out
@@ -138,25 +137,56 @@ def reconcile(token: str, secret: str) -> str:
 
 # --- 2. domain labels --------------------------------------------------------
 
+def team_writes_issues(team: dict) -> bool:
+    """Issue write from the team's unit map (Gitea >= 1.20), else its team-wide mode."""
+    if team.get("permission") in ("admin", "owner"):
+        return True
+    units = team.get("units_map") or {}
+    return (units.get(ISSUES_UNIT) if units else team.get("permission")) in BOT_WRITE
+
+
 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.
-    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"):
+    """Repos whose issues Hermes works: in each org BOT_LOGIN belongs to, the repos of its
+    teams that grant issue write (every org repo for an includes_all_repositories team),
+    issue tracker on, not archived. Not /collaborators/{user}/permission: that reports
+    the repo-wide access mode, which stays "none" when the write comes from a unit-based
+    team. The admin route lists private orgs too and needs only the PAT's admin scope
+    (/users/{u}/orgs also requires read:user). An org or team whose lookup fails is
+    skipped and named in `errors`."""
+    def failed(what: str, exc: RuntimeError) -> None:
+        if errors is None:
+            raise exc
+        errors.append(f"{what}: {exc}")
+
+    found: dict[str, dict] = {}
+    for org in paged(f"/admin/users/{q(BOT_LOGIN)}/orgs", token):
+        name = org.get("username") or org.get("name") or ""
+        if not name:
             continue
-        owner, _, name = r["full_name"].partition("/")
         try:
-            perm = api("GET", f"/repos/{q(owner)}/{q(name)}/collaborators/{q(BOT_LOGIN)}/permission", token)
+            teams = paged(f"/orgs/{q(name)}/teams", token)
         except RuntimeError as exc:
-            if errors is None:
-                raise
-            errors.append(f"{r['full_name']}: {exc}")
+            failed(name, exc)
             continue
-        if (perm or {}).get("permission") in BOT_WRITE:
-            out.append(r)
-    return out
+        for team in teams:
+            if not team_writes_issues(team):
+                continue
+            try:
+                try:
+                    api("GET", f"/teams/{team['id']}/members/{q(BOT_LOGIN)}", token)
+                except RuntimeError as exc:
+                    if str(exc).endswith("HTTP 404"):
+                        continue  # not a member of this team
+                    raise
+                repos = paged(f"/orgs/{q(name)}/repos" if team.get("includes_all_repositories")
+                              else f"/teams/{team['id']}/repos", token)
+            except RuntimeError as exc:
+                failed(f"{name} team {team.get('name')}", exc)
+                continue
+            for r in repos:
+                if r.get("has_issues", True) and not r.get("archived"):
+                    found.setdefault(r["full_name"], r)
+    return [found[k] for k in sorted(found)]
 
 
 def ensure_labels_at(path: str, token: str) -> list[str]:
@@ -170,18 +200,14 @@ def ensure_labels_at(path: str, token: str) -> list[str]:
 
 
 def ensure_labels(token: str, repos: list[dict]) -> str:
-    """Org labels in each org that owns one of `repos`, repo labels in the user-owned ones."""
-    orgs = {n for o in paged("/admin/orgs", token) if (n := o.get("username") or o.get("name"))}
-    owners = {(r.get("owner") or {}).get("login") for r in repos}
-    targets = [f"/orgs/{q(org)}/labels" for org in sorted(orgs & owners)]
-    targets += [f"/repos/{q(r['owner']['login'])}/{q(r['name'])}/labels" for r in repos
-                if (r.get("owner") or {}).get("login") not in orgs]
+    """Org labels in each org that owns one of `repos` (all org repos); never repo labels."""
+    orgs = sorted({o for r in repos if (o := (r.get("owner") or {}).get("login"))})
     actions = []
-    for path in targets:
-        created = ensure_labels_at(path, token)
+    for org in orgs:
+        created = ensure_labels_at(f"/orgs/{q(org)}/labels", token)
         if created:
-            actions.append(f"{path.rsplit('/labels', 1)[0]}: created {','.join(created)}")
-    return "; ".join(actions) or f"labels present in {len(targets)} orgs/repos"
+            actions.append(f"/orgs/{org}: created {','.join(created)}")
+    return "; ".join(actions) or f"labels present in {len(orgs)} orgs"
 
 
 # --- 3. backfill -------------------------------------------------------------
@@ -261,9 +287,9 @@ def main() -> int:
     def load_repos() -> str:
         errors: list[str] = []
         repos.extend(issue_repos(token, errors))
-        msg = f"{len(repos)} with issues writable by {BOT_LOGIN}"
+        msg = f"{len(repos)} org 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)}")
+            raise RuntimeError(f"{msg}; team lookup failed, skipped: {'; '.join(errors)}")
         return msg
 
     failures = 0

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

@@ -316,62 +316,120 @@ def repo(full, has_issues=True, archived=False):
             "has_issues": has_issues, "archived": archived}
 
 
+def team(tid, name, issues="write", all_repos=True, permission="write"):
+    return {"id": tid, "name": name, "includes_all_repositories": all_repos,
+            "permission": permission, "units_map": {"repo.issues": issues, "repo.code": "write"}}
+
+
+def fake_gitea(orgs, teams, members, repos, fail=()):
+    """api() stand-in: orgs of the bot, teams per org, bot's team ids, repos per org or team."""
+    seen = []
+
+    def fake(method, path, token, body=None):
+        p = path.split("?")[0]
+        seen.append(p)
+        if p in fail:
+            raise RuntimeError(f"GET {p}: HTTP 500")
+        if p == "/admin/users/hermes-bot/orgs":
+            return [{"username": o} for o in orgs]
+        if p.startswith("/orgs/") and p.endswith("/teams"):
+            return teams.get(p.split("/")[2], [])
+        if p.startswith("/teams/") and "/members/" in p:
+            if int(p.split("/")[2]) in members:
+                return {"login": "hermes-bot"}
+            raise RuntimeError(f"GET {p}: HTTP 404")
+        if p.endswith("/repos"):
+            return repos.get(p, [])
+        raise AssertionError(f"unexpected {method} {p}")
+
+    return fake, seen
+
+
 class DomainLabels(unittest.TestCase):
-    def test_labels_only_for_orgs_and_user_repos_hermes_works(self):
+    def test_org_labels_only_never_repo_labels(self):
         calls = []
 
         def fake(method, path, token, body=None):
             calls.append((method, path, body))
-            if path.startswith("/admin/orgs"):
-                return [{"username": "konairius"}, {"username": "unrelated-org"}]
             if path.startswith("/orgs/konairius/labels"):
                 return [{"name": "Atlas"}, {"name": "bug"}]
             return []
 
-        repos = [repo("konairius/homelab"), repo("k.renner/turm1388")]
+        repos = [repo("konairius/homelab"), repo("konairius/shelly-manager")]
         with mock.patch.object(sync, "api", side_effect=fake):
             msg = sync.ensure_labels("tok", repos)
         posts = [(p, b["name"]) for m, p, b in calls if m == "POST"]
         self.assertEqual(sorted(n for p, n in posts if p == "/orgs/konairius/labels"),
                          ["apollo", "clio", "daedalus", "hestia"])  # atlas exists (any case)
-        self.assertEqual(len([p for p, _ in posts if p == "/repos/k.renner/turm1388/labels"]), 5)
-        self.assertFalse([p for p, _ in posts if p.startswith("/repos/konairius/")])  # org labels reach it
-        self.assertFalse([c for c in calls if "unrelated-org" in c[1]])  # no repo of it is worked
+        self.assertFalse([c for c in calls if c[1].startswith("/repos/")])  # org labels reach them
+        self.assertFalse([c for c in calls if c[1].startswith("/admin/")])
         self.assertIn("created", msg)
 
-    def test_repos_without_issues_archived_or_bot_write_are_skipped(self):
-        perms = {"a/x": "write", "a/w": "admin", "gitea_admin/_cargo-index": "read", "a/n": "none"}
-        seen = []
-
-        def fake(method, path, token, body=None):
-            if path.startswith("/repos/search"):
-                return {"ok": True, "data": [
-                    repo("a/x"), repo("a/w"), repo("gitea_admin/_cargo-index"), repo("a/n"),
-                    repo("a/y", has_issues=False), repo("a/z", archived=True)]}
-            seen.append(path)
-            full = path.split("/repos/")[1].split("/collaborators/")[0]
-            return {"permission": perms[full]}
-
+    def test_team_write_covers_every_org_repo_without_collaborator(self):
+        # Live 2026-10-21: hermes-bot is only in team Bots (all repos, issues write);
+        # /collaborators/hermes-bot/permission says "none" for 9 of the 10 repos.
+        org_repos = [repo("konairius/homelab"), repo("konairius/shelly-manager"),
+                     repo("konairius/ical-proxy"), repo("konairius/off", has_issues=False),
+                     repo("konairius/old", archived=True)]
+        fake, seen = fake_gitea(
+            orgs=["konairius"],
+            teams={"konairius": [team(1, "Owners", "owner", permission="owner"), team(2, "Bots"),
+                                 team(5, "Reviewers", "read", permission="none")]},
+            members={2},
+            repos={"/orgs/konairius/repos": org_repos})
         with mock.patch.object(sync, "api", side_effect=fake):
             got = [r["full_name"] for r in sync.issue_repos("tok")]
-        self.assertEqual(got, ["a/x", "a/w"])
-        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/"))])
+        self.assertEqual(got, ["konairius/homelab", "konairius/ical-proxy", "konairius/shelly-manager"])
+        self.assertFalse([p for p in seen if "/collaborators/" in p or p.startswith("/repos/")])
+        self.assertNotIn("/teams/5/members/hermes-bot", seen)  # read-only team is not even checked
 
-    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"}
+    def test_team_with_selected_repos_and_no_issue_write(self):
+        fake, seen = fake_gitea(
+            orgs=["a", "b"],
+            teams={"a": [team(3, "Some", all_repos=False)],
+                   "b": [team(4, "CodeOnly", issues="read")]},
+            members={3, 4},
+            repos={"/teams/3/repos": [repo("a/x"), repo("a/y")],
+                   "/orgs/a/repos": [repo("a/x"), repo("a/y"), repo("a/not-in-team")],
+                   "/orgs/b/repos": [repo("b/z")]})
+        with mock.patch.object(sync, "api", side_effect=fake):
+            got = [r["full_name"] for r in sync.issue_repos("tok")]
+        self.assertEqual(got, ["a/x", "a/y"])
+        self.assertNotIn("/orgs/b/repos", seen)
 
+    def test_team_issue_write_from_units_map_or_team_mode(self):
+        self.assertTrue(sync.team_writes_issues(team(2, "Bots")))
+        self.assertFalse(sync.team_writes_issues(team(5, "Reviewers", "read", permission="none")))
+        self.assertTrue(sync.team_writes_issues(team(1, "Owners", "read", permission="owner")))
+        self.assertTrue(sync.team_writes_issues({"permission": "write"}))  # no units_map
+        self.assertFalse(sync.team_writes_issues({"permission": "read", "units_map": {}}))
+
+    def test_user_repos_are_out_of_scope(self):
+        fake, seen = fake_gitea(orgs=[], teams={}, members=set(), repos={})
+        with mock.patch.object(sync, "api", side_effect=fake):
+            self.assertEqual(sync.issue_repos("tok"), [])
+        self.assertEqual(seen, ["/admin/users/hermes-bot/orgs"])  # no /repos/search over user repos
+
+    def test_failed_team_lookup_skips_only_that_team(self):
+        fake, _ = fake_gitea(
+            orgs=["a", "broken"],
+            teams={"a": [team(3, "Bots", all_repos=False), team(6, "More", all_repos=False)]},
+            members={3, 6},
+            repos={"/teams/3/repos": [repo("a/x")], "/teams/6/repos": [repo("a/w"), repo("a/x")]},
+            fail={"/orgs/broken/teams", "/teams/6/members/hermes-bot"})
         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])
+        self.assertEqual(got, ["a/x"])
+        self.assertEqual(len(errors), 2)
+        self.assertIn("a team More", errors[0])
+        self.assertTrue(errors[1].startswith("broken:"))
+
+    def test_failed_lookup_raises_without_error_list(self):
+        fake, _ = fake_gitea(orgs=["broken"], teams={}, members=set(), repos={},
+                             fail={"/orgs/broken/teams"})
+        with mock.patch.object(sync, "api", side_effect=fake), self.assertRaises(RuntimeError):
+            sync.issue_repos("tok")
 
 
 class Backfill(unittest.TestCase):