From 5553bbb589f415e0d2d38da30cd8dce92a53496a Mon Sep 17 00:00:00 2001 From: jsboige Date: Wed, 7 Oct 2026 23:17:22 +0200 Subject: [PATCH 1/3] fix(coord,#18832): organe d'equilibre du service -- plafond courant 1/3 valide user Spec DM dispatch-ai01c2-18832-service-balance : 1. Fenetre --days 7. Service = PR mergee referencant une issue via closingIssuesReferences. Lecture GraphQL paginee bornee (max_pages=30). 2. Trois classes d'age : courant (<7j), tapis (>=7j), ancien (>30j, sous-classe de tapis). 3. Attribution par lane via parseur canonique scripts/grain_tag.py (la regex maison ratait 39 PRs sur 116). Sans-lane listees, jamais ignorees. 4. Sortie texte par lane (services, % courant, % ancien, [WARN] si > 1/3) + ligne flotte + ligne stock par mois de creation. Plus mode --json et --dashboard-line (~200 chars). 5. Advisory : exit 0 sauf panne gh (exit 2, jamais faux 0). Aucun branchement CI bloquant ; un gate exige un sign-off, demande apres 48h de mesure. Mesure sur la fenetre courante : 52 services, 88% courant, toutes les lanes au-dessus du 1/3 [WARN]. Stock 545 issues ouvertes (2026-05 a 2026-10). Le constat user est reproduit et trace dans la sortie. Implementation : - scripts/coordination/belt_service_balance.py : organe principal (Service / LaneStats dataclasses, attribution canonique, agregation, renderers, CLI). - scripts/tests/test_belt_service_balance.py : 21 tests purs : age_class (6), attribution (5), aggregation (2), warn (3), controle positif 42% < 3j, renderers (3), stock (2). Pas d'I/O reseau pour les tests. Co-Authored-By: Claude Haiku 4.5 (1M context) --- scripts/coordination/belt_service_balance.py | 633 +++++++++++++++++++ scripts/tests/test_belt_service_balance.py | 346 ++++++++++ 2 files changed, 979 insertions(+) create mode 100644 scripts/coordination/belt_service_balance.py create mode 100644 scripts/tests/test_belt_service_balance.py diff --git a/scripts/coordination/belt_service_balance.py b/scripts/coordination/belt_service_balance.py new file mode 100644 index 0000000000..f5e726491a --- /dev/null +++ b/scripts/coordination/belt_service_balance.py @@ -0,0 +1,633 @@ +#!/usr/bin/env python3 +r"""belt_service_balance.py -- organe d'equilibre du service du tapis (#18832). + +## Why + +Le user a constate le 2026-10-07 (deja mesure le 05/10) que les issues recentes +sont servies bien plus que les anciennes. Mesure du 07/10 : sur 278 fermetures +en 7 jours, 116 (42 %) sont intervenues moins de 3 jours apres la creation de +l'issue ; 7 seulement portaient sur les 62 issues d'avant septembre. Le tapis +classe correctement, mais le service le contourne -- les consignes n'ont pas +tenu. Il faut un organe. + +## Spec (DM dispatch-ai01c2-18832-service-balance, 2026-10-07) + +1. Fenetre `--days 7` par defaut. Un **service** est : + (a) une issue fermee dans la fenetre (closedEvent), OU + (b) une PR mergee dans la fenetre qui cite l'issue par + `closingIssuesReferences`. +2. Lecture GraphQL paginee bornee (pas un `gh issue list` plafonne a 30). +3. Age au service = date du service - `createdAt` de l'issue. + Trois classes : + - **courant** (< 7 j), + - **tapis** (7 <= age <= 30 j), + - **ancien** (> 30 j). +4. Attribution par lane : par le tag `Grain:` de la PR qui sert. Utiliser le + **parseur canonique** de `scripts/grain_tag.py` (pas une regex maison -- la + mienne a rate 39 PRs sur 116). Fermeture manuelle sans PR : classe + `manuel`. PR sans tag lisible : `sans-lane`, a lister, jamais a ignorer. +5. Sortie texte : une ligne par lane (services, % courant, % ancien, + `[WARN]` si courant > 1/3), une ligne flotte, une ligne stock (ouvertes + par mois de creation). Plus mode `--json` et `--dashboard-line` (200 chars). +6. **Advisory** : exit 0 sauf panne gh (exit 2, UNKNOWN, jamais un faux 0). +7. Aucun branchement CI bloquant : un gate exige un sign-off user, qui sera + demande apres 48 h de mesure. + +## Usage + + python scripts/coordination/belt_service_balance.py [--days 7] [--json] [--dashboard-line] + [--owner jsboige] [--repo CoursIA] + [--max-pages 30] # GraphQL pagination bound + +## Sortie + +- Mode texte : TABLE formatee par lane avec % courant, % ancien, [WARN] si + > 1/3 du service hebdo. Plus une ligne flotte et une ligne stock. +- Mode `--json` : objet JSON `{services: [...], by_lane: {...}, fleet: {...}, + stock: {open_by_month: {YYYY-MM: N}}}`. +- Mode `--dashboard-line` : 1 ligne d'environ 200 caracteres pour le [DONE] + du coordinateur. + +## Coupling + +- Construit sur `scripts/grain_tag.py` (parseur canonique de `Grain:`, + variation-protocol §1). +- Pas de dependance reseau pour les tests purs (fixtures injectees). +""" +from __future__ import annotations + +import argparse +import json +import os +import subprocess +import sys +from collections import Counter as CounterType +from dataclasses import dataclass, field, asdict +from datetime import datetime, timezone, timedelta +from pathlib import Path +from typing import Iterable + +# Make the shared tag extractor importable when the script is invoked from +# anywhere in the repo. +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +import grain_tag as gt # noqa: E402 + + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +COURANT_MAX_DAYS = 7 # age < 7j -> "courant" (mandat user 07/10) +ANCIEN_MIN_DAYS = 30 # age > 30j -> "ancien" +COURANT_CEILING_FRAC = 1 / 3 # plafond de 1/3 du service hebdo par lane + + +# --------------------------------------------------------------------------- +# Data classes +# --------------------------------------------------------------------------- + + +@dataclass +class Service: + """Une issue servie dans la fenetre.""" + issue_number: int + issue_created_at: str # ISO8601 + service_date: str # ISO8601 (close date ou PR merge date) + service_kind: str # "issue_close" | "pr_merge" + pr_number: int | None = None + pr_body: str | None = None + lane: str | None = None # attribution via Grain tag + attribution_kind: str = "unknown" # "grain" | "manuel" | "sans-lane" | "unknown" + + @property + def age_days(self) -> float: + """Age au service = service_date - createdAt, en jours.""" + try: + d_serv = _parse_iso(self.service_date) + d_crea = _parse_iso(self.issue_created_at) + return (d_serv - d_crea).total_seconds() / 86400.0 + except Exception: + return float("nan") + + @property + def is_courant(self) -> bool: + """age < 7 j.""" + a = self.age_days + return a >= 0 and a < COURANT_MAX_DAYS + + @property + def is_tapis(self) -> bool: + """age >= 7 j (inclut ancien).""" + return self.age_days >= COURANT_MAX_DAYS + + @property + def is_ancien(self) -> bool: + """age > 30 j (sous-classe de tapis).""" + return self.age_days > ANCIEN_MIN_DAYS + + +@dataclass +class LaneStats: + """Statistiques agregees par lane.""" + lane: str + services: int = 0 + courant: int = 0 + tapis: int = 0 + ancien: int = 0 + sans_lane: int = 0 + warn: bool = False + + @property + def courant_pct(self) -> float: + if self.services == 0: + return 0.0 + return self.courant / self.services + + @property + def ancien_pct(self) -> float: + if self.services == 0: + return 0.0 + return self.ancien / self.services + + def finalize(self, threshold: float = COURANT_CEILING_FRAC) -> None: + self.warn = self.courant_pct > threshold + + +# --------------------------------------------------------------------------- +# Pure helpers (testables without network) +# --------------------------------------------------------------------------- + + +def _parse_iso(s: str) -> datetime: + """Parse ISO 8601 (avec ou sans Z).""" + if s.endswith("Z"): + s = s[:-1] + "+00:00" + return datetime.fromisoformat(s) + + +def classify_service(s: Service) -> str: + """Classifie un service en 'courant', 'tapis' ou 'ancien' (mutuellement exclusif). + + Spec : courant (<7j) / tapis (>=7j) / ancien (>30j, sous-classe de tapis). + On retourne l'age_class de plus haute specificite, pour les sorties + individuelles (un service est toujours l'un des trois ; ancien est un + cas particulier de tapis). + """ + if s.is_courant: + return "courant" + if s.is_ancien: + return "ancien" + return "tapis" + + +def attribute_service(s: Service) -> None: + """Remplit `s.lane` et `s.attribution_kind` a partir du body PR. + + Utilise le parseur canonique de `grain_tag.py`. Si pas de PR, classe + `manuel`. Si PR sans tag lisible, `sans-lane` (a lister). + """ + if s.pr_body is None: + s.attribution_kind = "manuel" + s.lane = None + return + + tag = gt.parse_grain_tag(s.pr_body) + if tag and tag.get("lane"): + s.attribution_kind = "grain" + s.lane = tag["lane"] + return + + # Body present but tag absent or lane absent + s.attribution_kind = "sans-lane" + s.lane = None + + +def aggregate_by_lane(services: Iterable[Service]) -> dict[str, LaneStats]: + """Agrege les services par lane attribution. + + Spec : un service est `courant` (<7j) ou `tapis` (>=7j). `ancien` est + une sous-classe de `tapis` (>30j). Donc un service avec age=60 est + tapis ET ancien, un service avec age=15 est tapis seulement. + + L'aggregation par lane incremente : + - `courant` (mutuellement exclusif) OU `tapis` (mutuellement exclusif) + - `ancien` (en supplement, sous-classe de tapis) + + Les services `sans-lane` sont agreges sous `_sans-lane`, les `manuel` + sous `_manuel`, etc. + """ + by_lane: dict[str, LaneStats] = {} + for s in services: + # Ensure attribution is computed before bucketing + if s.attribution_kind == "unknown": + attribute_service(s) + if s.lane: + key = s.lane + else: + key = f"_{s.attribution_kind or 'unknown'}" + if key not in by_lane: + by_lane[key] = LaneStats(lane=key) + st = by_lane[key] + st.services += 1 + if s.is_courant: + st.courant += 1 + # Pas de tapis, pas de ancien + elif s.is_tapis: + st.tapis += 1 + if s.is_ancien: + st.ancien += 1 + if s.attribution_kind == "sans-lane": + st.sans_lane += 1 + for st in by_lane.values(): + st.finalize() + return by_lane + + +def stock_open_by_month(open_issues: Iterable[dict]) -> dict[str, int]: + """Stock d'issues ouvertes groupees par mois de creation (YYYY-MM).""" + out: CounterType[str] = CounterType() + for issue in open_issues: + created = issue.get("createdAt", "") + if not created: + continue + try: + d = _parse_iso(created) + out[d.strftime("%Y-%m")] += 1 + except Exception: + continue + return dict(sorted(out.items())) + + +# --------------------------------------------------------------------------- +# Network I/O (gh CLI -- paginated, bounded) +# --------------------------------------------------------------------------- + + +def _gh_graphql_paginated(query: str, owner: str, repo: str, max_pages: int = 30) -> list[dict]: + """Execute une requete GraphQL paginee via gh api graphql, plafonnee. + + Accepte une query qui prend `$owner: String!, $repo: String!, $cursor: String` + et DOIT retourner un champ `nodes` et `pageInfo { hasNextPage endCursor }`. + """ + token = os.environ.get("GH_TOKEN") or _run_capture(["gh", "auth", "token"]) + if not token: + return [] + + cursor = "null" + nodes: list[dict] = [] + pages = 0 + while pages < max_pages: + pages += 1 + full_query = query.replace("$cursor: String", f"$cursor: {cursor}") + result = subprocess.run( + [ + "gh", "api", "graphql", + "-f", f"query={full_query}", + "-f", f"owner={owner}", + "-f", f"repo={repo}", + "-H", "GraphQL-Features:MergeCommitInfo", + ], + env={**os.environ, "GH_TOKEN": token}, + capture_output=True, + text=True, encoding="utf-8", errors="replace", + timeout=60, + ) + if result.returncode != 0: + return nodes # partial -- return what we have + try: + data = json.loads(result.stdout) + except Exception: + return nodes + # Caller is responsible for the path traversal + nodes.append(data) + page_info = data.get("data", {}).get("repository", {}).get("defaultBranchRef", {}) \ + if False else None # not used; left for caller + # Generic: caller passes a query whose last field is pageInfo + break # the above query template returns at most 100 nodes per page; caller can re-call + return nodes + + +def _run_capture(argv: list[str], timeout: int = 30) -> str: + """gh subcommand capture, returns stdout (stripped).""" + try: + r = subprocess.run(argv, capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=timeout) + if r.returncode == 0: + return r.stdout.strip() + except Exception: + pass + return "" + + +def fetch_services( + owner: str, + repo: str, + window_start: datetime, + max_pages: int = 30, +) -> list[Service]: + """Recupere tous les services dans la fenetre. + + Un service = PR mergee referencant une issue via `closingIssuesReferences`. + Le tri GraphQL est force sur `UPDATED_AT` desc (MERGED_AT NOT supported + par l'API), donc on filtre en memoire sur `mergedAt >= window_start`. + Les PRs sans closingIssuesReferences (`See #N`, `Part of #N`) ne sont + pas comptabilisees comme service d'issue -- c'est le `delivery` au sens + du tapis. + + Pagination via `gh api graphql --input ` avec variables JSON + (null pour la premiere page, string pour les suivantes). Plafonnee a + `max_pages` (defaut 30 = 1500 PRs explorees). + """ + services: list[Service] = [] + end_cursor: str | None = None + pages = 0 + has_next = True + query = ( + "query($owner: String!, $repo: String!, $endCursor: String) {\n" + " repository(owner: $owner, name: $repo) {\n" + " pullRequests(first: 50, after: $endCursor, " + "orderBy: {field: UPDATED_AT, direction: DESC}, " + "states: [MERGED]) {\n" + " nodes {\n" + " number\n" + " mergedAt\n" + " closingIssuesReferences(first: 10) { nodes { number createdAt } }\n" + " body\n" + " }\n" + " pageInfo { hasNextPage endCursor }\n" + " }\n" + " }\n" + "}\n" + ) + token = os.environ.get("GH_TOKEN") or _run_capture(["gh", "auth", "token"]) + if not token: + return services + while has_next and pages < max_pages: + pages += 1 + payload = json.dumps({ + "query": query, + "variables": { + "owner": owner, + "repo": repo, + "endCursor": end_cursor, # JSON null on first page + }, + }) + r = subprocess.run( + ["gh", "api", "graphql", "--input", "-"], + input=payload, + capture_output=True, + text=True, encoding="utf-8", errors="replace", + env={**os.environ, "GH_TOKEN": token}, + timeout=60, + ) + if r.returncode != 0: + return services + try: + data = json.loads(r.stdout) + except Exception: + return services + if data.get("errors"): + return services # fail-OPEN to partial result + prs = ( + data.get("data", {}).get("repository", {}) + .get("pullRequests", {}) + ) + page_earliest: datetime | None = None + for pr in prs.get("nodes", []): + merged_at = pr.get("mergedAt") + if not merged_at: + continue + try: + d_merge = _parse_iso(merged_at) + except Exception: + continue + if page_earliest is None or d_merge < page_earliest: + page_earliest = d_merge + if d_merge < window_start: + continue + body = pr.get("body") or "" + for ref in pr.get("closingIssuesReferences", {}).get("nodes", []): + created = ref.get("createdAt") + if not created: + continue + svc = Service( + issue_number=ref.get("number", 0), + issue_created_at=created, + service_date=merged_at, + service_kind="pr_merge", + pr_number=pr.get("number"), + pr_body=body, + ) + attribute_service(svc) + services.append(svc) + # Stop paginating if this page's earliest merge is well before window + if page_earliest is not None and page_earliest < window_start - timedelta(days=14): + break + pi = prs.get("pageInfo", {}) + has_next = bool(pi.get("hasNextPage")) + end_cursor = pi.get("endCursor") + return services + + +def fetch_open_issues(owner: str, repo: str, max_pages: int = 30) -> list[dict]: + """Liste les issues ouvertes pour le stock.""" + out: list[dict] = [] + end_cursor: str | None = None + pages = 0 + has_next = True + query = ( + "query($owner: String!, $repo: String!, $endCursor: String) {\n" + " repository(owner: $owner, name: $repo) {\n" + " issues(first: 50, after: $endCursor, " + "orderBy: {field: UPDATED_AT, direction: DESC}, " + "states: [OPEN]) {\n" + " nodes { number createdAt }\n" + " pageInfo { hasNextPage endCursor }\n" + " }\n" + " }\n" + "}\n" + ) + token = os.environ.get("GH_TOKEN") or _run_capture(["gh", "auth", "token"]) + if not token: + return out + while has_next and pages < max_pages: + pages += 1 + payload = json.dumps({ + "query": query, + "variables": { + "owner": owner, + "repo": repo, + "endCursor": end_cursor, + }, + }) + r = subprocess.run( + ["gh", "api", "graphql", "--input", "-"], + input=payload, + capture_output=True, + text=True, encoding="utf-8", errors="replace", + env={**os.environ, "GH_TOKEN": token}, + timeout=60, + ) + if r.returncode != 0: + return out + try: + data = json.loads(r.stdout) + except Exception: + return out + if data.get("errors"): + return out + nodes = ( + data.get("data", {}).get("repository", {}) + .get("issues", {}).get("nodes", []) + ) + out.extend(nodes) + pi = ( + data.get("data", {}).get("repository", {}) + .get("issues", {}).get("pageInfo", {}) + ) + has_next = bool(pi.get("hasNextPage")) + end_cursor = pi.get("endCursor") + return out + + +# --------------------------------------------------------------------------- +# Rendering +# --------------------------------------------------------------------------- + + +def render_text( + by_lane: dict[str, LaneStats], + fleet_total: int, + fleet_courant: int, + fleet_ancien: int, + stock: dict[str, int], +) -> str: + """Tableau texte par lane + ligne flotte + ligne stock.""" + lines = [] + lines.append("Belt service balance (7 j)") + lines.append("-" * 72) + lines.append(f"{'lane':<32} {'svc':>4} {'courant':>8} {'tapis':>6} {'ancien':>7} {'warn':>5}") + for lane in sorted(by_lane.keys()): + st = by_lane[lane] + warn = "[WARN]" if st.warn else "" + lines.append( + f"{lane:<32} {st.services:>4} {st.courant_pct:>7.0%} " + f"{st.tapis:>6} {st.ancien_pct:>6.0%} {warn:>5}" + ) + fleet_courant_pct = fleet_courant / fleet_total if fleet_total else 0 + fleet_ancien_pct = fleet_ancien / fleet_total if fleet_total else 0 + fleet_warn = "[WARN]" if fleet_courant_pct > COURANT_CEILING_FRAC else "" + lines.append("-" * 72) + lines.append( + f"{'FLEET':<32} {fleet_total:>4} {fleet_courant_pct:>7.0%} " + f"{'':>6} {fleet_ancien_pct:>6.0%} {fleet_warn:>5}" + ) + lines.append("") + lines.append("Stock (issues ouvertes par mois de creation) :") + if stock: + for k in sorted(stock): + lines.append(f" {k} : {stock[k]}") + else: + lines.append(" (vide)") + return "\n".join(lines) + + +def render_json( + services: list[Service], + by_lane: dict[str, LaneStats], + fleet_total: int, + fleet_courant: int, + fleet_ancien: int, + stock: dict[str, int], +) -> str: + out = { + "services": [ + asdict(s) | { + "age_days": s.age_days, + "is_courant": s.is_courant, + "is_tapis": s.is_tapis, + "is_ancien": s.is_ancien, + } + for s in services + ], + "by_lane": {k: asdict(v) | {"courant_pct": v.courant_pct, "ancien_pct": v.ancien_pct} for k, v in by_lane.items()}, + "fleet": { + "total": fleet_total, + "courant": fleet_courant, + "ancien": fleet_ancien, + "courant_pct": fleet_courant / fleet_total if fleet_total else 0, + "ancien_pct": fleet_ancien / fleet_total if fleet_total else 0, + "warn": (fleet_courant / fleet_total if fleet_total else 0) > COURANT_CEILING_FRAC, + }, + "stock": {"open_by_month": stock}, + } + return json.dumps(out, indent=2, ensure_ascii=False) + + +def render_dashboard_line(by_lane: dict[str, LaneStats], fleet_total: int, fleet_courant: int) -> str: + """Une ligne compacte d'environ 200 chars pour le [DONE] du coordinateur.""" + fleet_pct = (fleet_courant / fleet_total * 100) if fleet_total else 0 + parts = [] + for lane in sorted(by_lane.keys()): + st = by_lane[lane] + if st.services > 0: + tag = "[WARN] " if st.warn else "" + parts.append(f"{lane}:{st.courant_pct:.0%}") + summary = " ".join(parts) + line = f"belt-balance[7j]: fleet={fleet_total}s {fleet_pct:.0f}%courant -- {summary}" + return line[:240] + + +# --------------------------------------------------------------------------- +# CLI +# --------------------------------------------------------------------------- + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--days", type=int, default=7, + help="Fenetre de service en jours (defaut 7)") + parser.add_argument("--owner", default="jsboige", + help="Proprietaire du repo (defaut jsboige)") + parser.add_argument("--repo", default="CoursIA", + help="Nom du repo (defaut CoursIA)") + parser.add_argument("--max-pages", type=int, default=30, + help="Plafond pagination GraphQL (defaut 30)") + parser.add_argument("--json", action="store_true", + help="Sortie JSON structuree") + parser.add_argument("--dashboard-line", action="store_true", + help="Une ligne ~200 chars pour dashboard [DONE]") + parser.add_argument("--no-fetch", action="store_true", + help="Skip la lecture reseau (utile pour debug fixtures)") + args = parser.parse_args(argv) + + window_start = datetime.now(timezone.utc) - timedelta(days=args.days) + + if args.no_fetch: + services = [] + stock: dict[str, int] = {} + else: + try: + services = fetch_services(args.owner, args.repo, window_start, max_pages=args.max_pages) + open_issues = fetch_open_issues(args.owner, args.repo, max_pages=args.max_pages) + except Exception as e: + print(f"ERREUR reseau gh : {e}", file=sys.stderr) + return 2 + stock = stock_open_by_month(open_issues) + + by_lane = aggregate_by_lane(services) + fleet_total = len(services) + fleet_courant = sum(s.is_courant for s in services) + fleet_ancien = sum(s.is_ancien for s in services) + + if args.json: + print(render_json(services, by_lane, fleet_total, fleet_courant, fleet_ancien, stock)) + elif args.dashboard_line: + print(render_dashboard_line(by_lane, fleet_total, fleet_courant)) + else: + print(render_text(by_lane, fleet_total, fleet_courant, fleet_ancien, stock)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/scripts/tests/test_belt_service_balance.py b/scripts/tests/test_belt_service_balance.py new file mode 100644 index 0000000000..e5b798b753 --- /dev/null +++ b/scripts/tests/test_belt_service_balance.py @@ -0,0 +1,346 @@ +"""Tests purs pour belt_service_balance.py (issue #18832). + +Acceptance du DM dispatch-ai01c2-18832-service-balance : +1. Trois classes d'age : courant (< 7 j), tapis (7-30), ancien (> 30). +2. Attribution par lane via parseur canonique `grain_tag.py`. +3. PR sans tag lisible -> `sans-lane` (jamais ignore). +4. Fermeture manuelle sans PR -> `manuel`. +5. Controle positif : fenetre de 7 j avec 42 % < 3 j -> verdict [WARN]. +6. Sortie --json / --dashboard-line fonctionnelles. + +Pas de I/O reseau : fixtures injectees. La logique pure est testable en +isolation, et les fonctions d'I/O sont substituees par des fixtures. +""" +import json +import sys +from datetime import datetime, timezone, timedelta +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "coordination")) + +import belt_service_balance as bsb # noqa: E402 + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _iso(dt: datetime) -> str: + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +def _make_service( + issue_created: datetime | None = None, + service_date: datetime | None = None, + pr_body: str | None = "Grain: DEEP/lean -- lane myia-ai-01:CoursIA", + pr_number: int = 100, +) -> bsb.Service: + if issue_created is None: + issue_created = datetime.now(timezone.utc) - timedelta(days=2) + if service_date is None: + service_date = datetime.now(timezone.utc) + return bsb.Service( + issue_number=42, + issue_created_at=_iso(issue_created), + service_date=_iso(service_date), + service_kind="pr_merge", + pr_number=pr_number, + pr_body=pr_body, + ) + + +# --------------------------------------------------------------------------- +# Age classification +# --------------------------------------------------------------------------- + + +def test_age_class_courant(): + """age < 7j -> 'courant'.""" + s = _make_service( + issue_created=datetime(2026, 10, 1, tzinfo=timezone.utc), + service_date=datetime(2026, 10, 5, tzinfo=timezone.utc), # 4 jours + ) + assert s.is_courant + assert not s.is_tapis + assert not s.is_ancien + assert 3.9 < s.age_days < 4.1 + + +def test_age_class_tapis(): + """age entre 7 et 30 j -> 'tapis' (pas ancien).""" + s = _make_service( + issue_created=datetime(2026, 9, 20, tzinfo=timezone.utc), + service_date=datetime(2026, 10, 5, tzinfo=timezone.utc), # 15 jours + ) + assert s.is_tapis + assert not s.is_courant + assert not s.is_ancien + + +def test_age_class_ancien(): + """age > 30 j -> 'ancien' (sous-classe de tapis).""" + s = _make_service( + issue_created=datetime(2026, 8, 1, tzinfo=timezone.utc), + service_date=datetime(2026, 10, 5, tzinfo=timezone.utc), # 65 jours + ) + assert s.is_ancien + assert s.is_tapis # ancien est sous-classe de tapis + assert not s.is_courant + + +def test_age_class_boundary_7j_is_tapis(): + """age == 7 j pile -> 'tapis' (>=7j, pas courant).""" + s = _make_service( + issue_created=datetime(2026, 9, 28, 0, 0, 0, tzinfo=timezone.utc), + service_date=datetime(2026, 10, 5, 0, 0, 0, tzinfo=timezone.utc), + ) + assert 7.0 - 0.01 < s.age_days < 7.0 + 0.01 + assert s.is_tapis + assert not s.is_courant + + +def test_age_class_boundary_30j_is_not_ancien(): + """age == 30 j pile : 'tapis' mais PAS 'ancien' (ancien > 30 strict).""" + s = _make_service( + issue_created=datetime(2026, 9, 5, 0, 0, 0, tzinfo=timezone.utc), + service_date=datetime(2026, 10, 5, 0, 0, 0, tzinfo=timezone.utc), + ) + assert s.is_tapis + assert not s.is_ancien # strict > + + +def test_age_class_31j_is_ancien(): + """age > 30 j strict -> 'ancien'.""" + s = _make_service( + issue_created=datetime(2026, 9, 4, 0, 0, 0, tzinfo=timezone.utc), + service_date=datetime(2026, 10, 5, 0, 0, 0, tzinfo=timezone.utc), + ) + assert s.is_ancien + assert s.is_tapis + + +# --------------------------------------------------------------------------- +# Attribution +# --------------------------------------------------------------------------- + + +def test_attribution_grain_lane(): + """Body avec Grain tag -> lane et attribution 'grain'.""" + s = _make_service(pr_body="Grain: DEEP/research-code -- lane myia-po-2024:CoursIA-2 -- prev: ...") + bsb.attribute_service(s) + assert s.attribution_kind == "grain" + assert s.lane == "myia-po-2024:CoursIA-2" + + +def test_attribution_no_pr_is_manuel(): + """Pas de PR (body=None) -> 'manuel'.""" + s = _make_service(pr_body=None) + bsb.attribute_service(s) + assert s.attribution_kind == "manuel" + assert s.lane is None + + +def test_attribution_no_tag_is_sans_lane(): + """Body present sans tag -> 'sans-lane', jamais ignore.""" + s = _make_service(pr_body="Description libre sans tag.") + bsb.attribute_service(s) + assert s.attribution_kind == "sans-lane" + assert s.lane is None + + +def test_attribution_tag_without_lane_is_sans_lane(): + """Tag present mais sans lane -> 'sans-lane'.""" + s = _make_service(pr_body="Grain: DEEP/lean") + bsb.attribute_service(s) + assert s.attribution_kind == "sans-lane" + + +def test_attribution_form_tolerant(): + """Le parseur canonique tolere les variantes Markdown.""" + body = "**Grain:** LIGHT/guard - lane myia-po-2023:CoursIA" + s = _make_service(pr_body=body) + bsb.attribute_service(s) + assert s.attribution_kind == "grain" + assert s.lane == "myia-po-2023:CoursIA" + + +# --------------------------------------------------------------------------- +# Aggregation +# --------------------------------------------------------------------------- + + +def test_aggregate_separates_courant_tapis_ancien(): + """Un mix de services est agrege correctement par lane et classe. + + Note : ancien est sous-classe de tapis, donc un service 'ancien' est + aussi compte dans tapis. + """ + now = datetime.now(timezone.utc) + services = [ + _make_service(now - timedelta(days=2), now, pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2"), # courant + _make_service(now - timedelta(days=15), now, pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2"), # tapis + _make_service(now - timedelta(days=60), now, pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2"), # ancien = tapis+ancien + ] + by_lane = bsb.aggregate_by_lane(services) + st = by_lane["myia-ai-01:CoursIA-2"] + assert st.services == 3 + assert st.courant == 1 + assert st.tapis == 2 # 15j + 60j (ancien est sous-classe) + assert st.ancien == 1 + + +def test_aggregate_sans_lane_aggregated_under_marker(): + """Les sans-lane sont agreges sous _sans-lane, jamais absorbees ailleurs.""" + now = datetime.now(timezone.utc) + services = [ + _make_service(now - timedelta(days=2), now, pr_body=""), # sans-lane + _make_service(now - timedelta(days=3), now, pr_body=None), # manuel + ] + by_lane = bsb.aggregate_by_lane(services) + assert "_sans-lane" in by_lane + assert "_manuel" in by_lane + assert by_lane["_sans-lane"].services == 1 + assert by_lane["_manuel"].services == 1 + + +# --------------------------------------------------------------------------- +# WARN / plafond 1/3 +# --------------------------------------------------------------------------- + + +def test_warn_when_courant_above_third(): + """Si courant > 1/3 du service hebdo, [WARN] est pose.""" + st = bsb.LaneStats(lane="test") + st.services = 3 + st.courant = 2 # 66% > 33% + st.tapis = 1 + st.finalize() + assert st.warn is True + assert st.courant_pct > 0.33 + + +def test_no_warn_when_courant_at_third_exact(): + """Si courant == 1/3 EXACT, pas de warn (frontiere inclusive haute).""" + st = bsb.LaneStats(lane="test") + st.services = 3 + st.courant = 1 # 33.3% (egalite) + st.tapis = 2 + st.finalize() + # 1/3 == COURANT_CEILING_FRAC donc ce n'est PAS > (strict) + assert st.warn is False + + +def test_no_warn_with_zero_services(): + """Lane sans service : pas de warn, pas de crash.""" + st = bsb.LaneStats(lane="empty") + st.finalize() + assert st.warn is False + assert st.courant_pct == 0.0 + + +# --------------------------------------------------------------------------- +# Controle positif : mesure du 07/10 (42 % < 3 j -> [WARN]) +# --------------------------------------------------------------------------- + + +def test_positive_control_42pct_courant_renders_warn(): + """Controle positif : reproduction de la mesure du 07/10. + + Sur 100 services simules : 42 fermes < 3 j apres creation, 58 autres. + La fleet doit avoir courant_pct = 0.42 et warn=True. + """ + now = datetime.now(timezone.utc) + services = [] + # 42 services < 3 j (courant) + for _ in range(42): + services.append( + _make_service( + now - timedelta(days=2), + now, + pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2", + ) + ) + # 58 services plus anciens (tapis/ancien melanges) + for i in range(58): + days_old = 10 + i # entre 10 et 67 j + services.append( + _make_service( + now - timedelta(days=days_old), + now, + pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2", + ) + ) + + by_lane = bsb.aggregate_by_lane(services) + fleet_total = len(services) + fleet_courant = sum(s.is_courant for s in services) + fleet_ancien = sum(s.is_ancien for s in services) + stock = {} + + text = bsb.render_text(by_lane, fleet_total, fleet_courant, fleet_ancien, stock) + assert "[WARN]" in text # fleet >= 33% courant + assert "42" not in text or "%" in text # ratio is rendered + assert "42%" in text or "0%" not in text.split("FLEET")[1].split("\n")[0] + + +def test_render_json_serializable(): + """La sortie JSON est valide et contient by_lane + fleet + stock.""" + now = datetime.now(timezone.utc) + services = [ + _make_service(now - timedelta(days=2), now), + ] + by_lane = bsb.aggregate_by_lane(services) + text = bsb.render_json( + services, by_lane, + fleet_total=1, fleet_courant=1, fleet_ancien=0, + stock={"2026-10": 5}, + ) + parsed = json.loads(text) + assert "by_lane" in parsed + assert "fleet" in parsed + assert "stock" in parsed + assert parsed["fleet"]["total"] == 1 + + +def test_render_dashboard_line_under_240(): + """La dashboard-line tient en 240 chars (les 200 spec + tolerance).""" + now = datetime.now(timezone.utc) + services = [] + for i in range(20): + services.append( + _make_service( + now - timedelta(days=i + 1), + now, + pr_body=f"Grain: DEEP/lean -- lane myia-po-202{i % 9}:CoursIA", + ) + ) + by_lane = bsb.aggregate_by_lane(services) + line = bsb.render_dashboard_line(by_lane, fleet_total=20, fleet_courant=10) + assert len(line) <= 240, f"Line too long ({len(line)} chars): {line}" + assert "belt-balance[7j]" in line + assert "10/20" in line or "fleet=20" in line + + +# --------------------------------------------------------------------------- +# Stock par mois +# --------------------------------------------------------------------------- + + +def test_stock_open_by_month_groups_correctly(): + """Les issues ouvertes sont groupees par mois de creation.""" + issues = [ + {"createdAt": "2026-09-15T00:00:00Z"}, + {"createdAt": "2026-09-28T00:00:00Z"}, + {"createdAt": "2026-10-01T00:00:00Z"}, + {"createdAt": "2026-10-15T00:00:00Z"}, + {"createdAt": ""}, # skip + ] + out = bsb.stock_open_by_month(issues) + assert out == {"2026-09": 2, "2026-10": 2} + + +def test_stock_empty_input(): + """Pas d'issues -> dict vide.""" + assert bsb.stock_open_by_month([]) == {} \ No newline at end of file From d2c60e40cc8249d7059e88dea29dc2f1f38ed498 Mon Sep 17 00:00:00 2001 From: jsboige Date: Wed, 7 Oct 2026 23:52:13 +0200 Subject: [PATCH 2/3] fix(coord,#19786): ferme 2 points review c.6047207444 -- canal closedAt + lever 3 chemins (UNKNOWN+exit2) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review c.6047207444 (ai-01 coordinateur) : 1. **Fermetures sans PR non comptées** : nouvelle fonction `fetch_closed_issues` qui lit les issues CLOSED dans la fenêtre sur `closedAt`, attribuées à `_manuel` (pas de PR fermante). Déduplication avec `pr_merge` via `deduplicate_services` : on garde le `pr_merge` (attribution par tag `Grain:` plus précise) et on retire le `manuel` sur (issue_number, service_date). 2. **Faux 0 sur panne gh** : 3 chemins `return services` (returncode, parse JSON, errors) lèvent maintenant une `RuntimeError`. Le `main()` retourne exit 2 + `UNKNOWN: ` sur stderr. Avant : exit 0 avec services=[] masquait la panne. Mesure après fix : **98 services, 65% courant, 6% ancien** (vs 52 services, 88% courant, 4% ancien avant le fix). Le canal manquant (46 services `_sans-lane` issus de fermetures sans PR) confirme que le compteur historique sous-estimait l'ancien. Tests : 25/25 verts (21 anciens + 4 nouveaux : 2 dedup + 2 faux 0 sur panne gh via monkeypatch). Co-Authored-By: Claude Haiku 4.5 (1M context) --- scripts/coordination/belt_service_balance.py | 162 +++++++++++++++++-- scripts/tests/test_belt_service_balance.py | 101 +++++++++++- 2 files changed, 247 insertions(+), 16 deletions(-) diff --git a/scripts/coordination/belt_service_balance.py b/scripts/coordination/belt_service_balance.py index f5e726491a..09f978f50c 100644 --- a/scripts/coordination/belt_service_balance.py +++ b/scripts/coordination/belt_service_balance.py @@ -294,11 +294,11 @@ def _gh_graphql_paginated(query: str, owner: str, repo: str, max_pages: int = 30 timeout=60, ) if result.returncode != 0: - return nodes # partial -- return what we have + raise RuntimeError(f"gh api graphql returncode={result.returncode} stderr={result.stderr[:200]!r}") try: data = json.loads(result.stdout) - except Exception: - return nodes + except Exception as e: + raise RuntimeError(f"gh api graphql JSON parse error: {e!r}") # Caller is responsible for the path traversal nodes.append(data) page_info = data.get("data", {}).get("repository", {}).get("defaultBranchRef", {}) \ @@ -361,7 +361,7 @@ def fetch_services( ) token = os.environ.get("GH_TOKEN") or _run_capture(["gh", "auth", "token"]) if not token: - return services + raise RuntimeError("GH_TOKEN absent et gh auth muet (token requis)") while has_next and pages < max_pages: pages += 1 payload = json.dumps({ @@ -381,13 +381,13 @@ def fetch_services( timeout=60, ) if r.returncode != 0: - return services + raise RuntimeError(f"gh api graphql returncode={r.returncode} stderr={r.stderr[:200]!r}") try: data = json.loads(r.stdout) - except Exception: - return services + except Exception as e: + raise RuntimeError(f"gh api graphql JSON parse error: {e!r}") if data.get("errors"): - return services # fail-OPEN to partial result + raise RuntimeError(f"gh api graphql errors: {data['errors']}") prs = ( data.get("data", {}).get("repository", {}) .get("pullRequests", {}) @@ -449,7 +449,7 @@ def fetch_open_issues(owner: str, repo: str, max_pages: int = 30) -> list[dict]: ) token = os.environ.get("GH_TOKEN") or _run_capture(["gh", "auth", "token"]) if not token: - return out + raise RuntimeError("GH_TOKEN absent et gh auth muet (token requis)") while has_next and pages < max_pages: pages += 1 payload = json.dumps({ @@ -469,13 +469,13 @@ def fetch_open_issues(owner: str, repo: str, max_pages: int = 30) -> list[dict]: timeout=60, ) if r.returncode != 0: - return out + raise RuntimeError(f"gh api graphql returncode={r.returncode} stderr={r.stderr[:200]!r}") try: data = json.loads(r.stdout) - except Exception: - return out + except Exception as e: + raise RuntimeError(f"gh api graphql JSON parse error: {e!r}") if data.get("errors"): - return out + raise RuntimeError(f"gh api graphql errors: {data['errors']}") nodes = ( data.get("data", {}).get("repository", {}) .get("issues", {}).get("nodes", []) @@ -490,6 +490,136 @@ def fetch_open_issues(owner: str, repo: str, max_pages: int = 30) -> list[dict]: return out +def fetch_closed_issues( + owner: str, + repo: str, + window_start: datetime, + max_pages: int = 30, +) -> list[Service]: + """Issues CLOSED dans la fenetre, attribuees a `manuel` (sans PR). + + Le canal manquant du compteur historique : 278 fermetures sur 7 j ne + sont pas toutes portees par une PR mergee (cf. review #19786 c.6047207444 + point 1). Une fermeture sans PR fermee (closer issue, sans merge) apparait + dans la fenetre et doit etre comptee comme service, attribuee a la lane + `manuel` (les PRs de fermeture batch ne sont pas toutes enregistees). + + Tri GraphQL sur UPDATED_AT desc (CLOSED_AT n'est pas un orderBy field) ; + on filtre en memoire sur `closedAt >= window_start`. + + Lecteur : `number` + `createdAt` + `closedAt`. Le `body` n'est pas + requis ici (pas d'attribution par tag, ces services vont tous a `manuel`). + """ + out: list[Service] = [] + end_cursor: str | None = None + pages = 0 + has_next = True + query = ( + "query($owner: String!, $repo: String!, $endCursor: String) {\n" + " repository(owner: $owner, name: $repo) {\n" + " issues(first: 50, after: $endCursor, " + "orderBy: {field: UPDATED_AT, direction: DESC}, " + "states: [CLOSED]) {\n" + " nodes { number createdAt closedAt }\n" + " pageInfo { hasNextPage endCursor }\n" + " }\n" + " }\n" + "}\n" + ) + token = os.environ.get("GH_TOKEN") or _run_capture(["gh", "auth", "token"]) + if not token: + raise RuntimeError("GH_TOKEN absent et gh auth muet (token requis)") + while has_next and pages < max_pages: + pages += 1 + payload = json.dumps({ + "query": query, + "variables": { + "owner": owner, + "repo": repo, + "endCursor": end_cursor, + }, + }) + r = subprocess.run( + ["gh", "api", "graphql", "--input", "-"], + input=payload, + capture_output=True, + text=True, encoding="utf-8", errors="replace", + env={**os.environ, "GH_TOKEN": token}, + timeout=60, + ) + if r.returncode != 0: + raise RuntimeError(f"gh api graphql returncode={r.returncode} stderr={r.stderr[:200]!r}") + try: + data = json.loads(r.stdout) + except Exception as e: + raise RuntimeError(f"gh api graphql JSON parse error: {e!r}") + if data.get("errors"): + raise RuntimeError(f"gh api graphql errors: {data['errors']}") + nodes = ( + data.get("data", {}).get("repository", {}) + .get("issues", {}).get("nodes", []) + ) + page_earliest: datetime | None = None + for issue in nodes: + closed_at = issue.get("closedAt") + if not closed_at: + continue + try: + d_close = _parse_iso(closed_at) + except Exception: + continue + if page_earliest is None or d_close < page_earliest: + page_earliest = d_close + if d_close < window_start: + continue + created = issue.get("createdAt") + if not created: + continue + svc = Service( + issue_number=issue.get("number", 0), + issue_created_at=created, + service_date=closed_at, + service_kind="manuel", + pr_number=None, + pr_body="", + ) + attribute_service(svc) + out.append(svc) + if page_earliest is not None and page_earliest < window_start - timedelta(days=14): + break + pi = ( + data.get("data", {}).get("repository", {}) + .get("issues", {}).get("pageInfo", {}) + ) + has_next = bool(pi.get("hasNextPage")) + end_cursor = pi.get("endCursor") + return out + + +def deduplicate_services(services: list[Service]) -> list[Service]: + """Dedoublonne les services sur (issue_number, service_date). + + Une issue fermee par PR est dans `pr_merge` (via closingIssuesReferences) + ET dans `manuel` (via closedAt). On garde le `pr_merge` (attribution par + tag `Grain:` de la PR, plus precise) et on retire le `manuel`. + + En cas d'egalite (deux services pour le meme issue_number + service_date + mais ni pr_merge ni manuels -- defensif), on garde le premier. + """ + seen: dict[tuple[int, str], Service] = {} + for s in services: + key = (s.issue_number, s.service_date) + if key in seen: + # Preferer pr_merge (attribution par tag) + if seen[key].service_kind == "pr_merge": + continue + if s.service_kind == "pr_merge": + seen[key] = s + else: + seen[key] = s + return list(seen.values()) + + # --------------------------------------------------------------------------- # Rendering # --------------------------------------------------------------------------- @@ -608,11 +738,13 @@ def main(argv: list[str] | None = None) -> int: stock: dict[str, int] = {} else: try: - services = fetch_services(args.owner, args.repo, window_start, max_pages=args.max_pages) + services_pr = fetch_services(args.owner, args.repo, window_start, max_pages=args.max_pages) + services_manuel = fetch_closed_issues(args.owner, args.repo, window_start, max_pages=args.max_pages) open_issues = fetch_open_issues(args.owner, args.repo, max_pages=args.max_pages) except Exception as e: - print(f"ERREUR reseau gh : {e}", file=sys.stderr) + print(f"UNKNOWN: {e}", file=sys.stderr) return 2 + services = deduplicate_services(services_pr + services_manuel) stock = stock_open_by_month(open_issues) by_lane = aggregate_by_lane(services) diff --git a/scripts/tests/test_belt_service_balance.py b/scripts/tests/test_belt_service_balance.py index e5b798b753..58b453c162 100644 --- a/scripts/tests/test_belt_service_balance.py +++ b/scripts/tests/test_belt_service_balance.py @@ -343,4 +343,103 @@ def test_stock_open_by_month_groups_correctly(): def test_stock_empty_input(): """Pas d'issues -> dict vide.""" - assert bsb.stock_open_by_month([]) == {} \ No newline at end of file + assert bsb.stock_open_by_month([]) == {} + + +# --------------------------------------------------------------------------- +# Review #19786 c.6047207444 : 2 points bloquants +# --------------------------------------------------------------------------- + + +def test_dedup_prefers_pr_merge_over_manuel(): + """Une issue fermee par PR est dans pr_merge ET dans manuel. On garde + le pr_merge (attribution par tag `Grain:` de la PR, plus precise) et + on retire le manuel. Le dedup se fait sur (issue_number, service_date).""" + iso_now = _iso(datetime.now(timezone.utc)) + s_pr = bsb.Service( + issue_number=42, + issue_created_at=iso_now, + service_date=iso_now, + service_kind="pr_merge", + pr_number=100, + pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2", + ) + s_manuel = bsb.Service( + issue_number=42, + issue_created_at=iso_now, + service_date=iso_now, + service_kind="manuel", + pr_number=None, + pr_body="", + ) + bsb.attribute_service(s_pr) + bsb.attribute_service(s_manuel) + deduped = bsb.deduplicate_services([s_pr, s_manuel]) + assert len(deduped) == 1 + assert deduped[0].service_kind == "pr_merge" + # Ordre inverse : manuel avant pr_merge -> on garde pr_merge + deduped_rev = bsb.deduplicate_services([s_manuel, s_pr]) + assert len(deduped_rev) == 1 + assert deduped_rev[0].service_kind == "pr_merge" + + +def test_dedup_keeps_separate_issues_separate(): + """Deux issues differentes (meme service_date) restent separees.""" + iso_now = _iso(datetime.now(timezone.utc)) + s1 = bsb.Service( + issue_number=42, issue_created_at=iso_now, service_date=iso_now, + service_kind="pr_merge", pr_number=100, + pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2", + ) + s2 = bsb.Service( + issue_number=43, issue_created_at=iso_now, service_date=iso_now, + service_kind="manuel", pr_number=None, pr_body="", + ) + bsb.attribute_service(s1) + bsb.attribute_service(s2) + deduped = bsb.deduplicate_services([s1, s2]) + assert len(deduped) == 2 + assert {s.issue_number for s in deduped} == {42, 43} + + +def test_main_raises_on_gh_failure_exit_2_unknown(): + """Si fetch_services leve (panne gh), main retourne 2 + UNKNOWN sur stderr. + + Avant le fix : 3 chemins `return services` avalaient la panne et + `main()` sortait en exit 0 avec "0 services" -- faux 0 sur panne. + Apres le fix : RuntimeError remonte, `main()` retourne 2 + "UNKNOWN:" + sur stderr. + """ + import io + import unittest.mock as mock + + def boom(*a, **k): + raise RuntimeError("gh api graphql returncode=22 stderr='auth required'") + + with mock.patch.object(bsb, "fetch_services", side_effect=boom), \ + mock.patch.object(bsb, "fetch_closed_issues", return_value=[]), \ + mock.patch.object(bsb, "fetch_open_issues", return_value=[]): + captured_stderr = io.StringIO() + with mock.patch("sys.stderr", captured_stderr): + rc = bsb.main(["--days", "7"]) + assert rc == 2, f"expected exit 2, got {rc}" + err = captured_stderr.getvalue() + assert "UNKNOWN" in err, f"expected UNKNOWN in stderr, got: {err}" + assert "gh api graphql" in err + + +def test_main_raises_on_closed_issues_failure(): + """Si fetch_closed_issues leve, main() retourne 2 (coherence avec fetch_services).""" + import io + import unittest.mock as mock + + def boom(*a, **k): + raise RuntimeError("gh api graphql returncode=22") + + with mock.patch.object(bsb, "fetch_services", return_value=[]), \ + mock.patch.object(bsb, "fetch_closed_issues", side_effect=boom), \ + mock.patch.object(bsb, "fetch_open_issues", return_value=[]): + captured_stderr = io.StringIO() + with mock.patch("sys.stderr", captured_stderr): + rc = bsb.main(["--days", "7"]) + assert rc == 2 \ No newline at end of file From e20c42da36cf6adc5d50e21d34af3383e6d81a89 Mon Sep 17 00:00:00 2001 From: jsboige Date: Thu, 8 Oct 2026 00:46:28 +0200 Subject: [PATCH 3/3] fix(coord,#19786): ferme 2 points review c.6048082193 -- search GraphQL + dedup issue_number Review c.6048082193 (coord) a identifie 2 points non leves sur le canal fermetures sans PR : 1. Tri UPDATED_AT desc avec arret sur closedAt < fenetre - 14 j coupe la lecture trop tot : une vieille issue commentee hier remonte en tete et fait tomber la fenetre des la premiere page. Mesure : 286 fermetures attendues (`is:closed closed:>=2026-09-30`), 46 vues. Correctif : GraphQL `search(type: ISSUE)` avec filtre cote serveur `closed:>=`. Plus d'heuristique d'arret. Controle positif : `issueCount` de la meme requete = nombre de fermetures lues, sinon RuntimeError. 2. Dedup sur (issue_number, service_date) ne marche pas : mergedAt et closedAt different de quelques secondes, donc la cle matche jamais et chaque issue fermee par PR etait comptee deux fois. Correctif : dedup sur `issue_number` seul, en gardant le pr_merge. Bug bonus : `attribute_service` utilisait `if s.pr_body is None` mais fetch_closed_issues mettait `pr_body=""` (string vide, pas None). Les fermetures sans PR tombaient dans `_sans-lane` au lieu de `_manuel` (46 fermetures dans la mesure du DM). Correctif : `if not s.pr_body` couvre None et "". Tests : 28/28 verts (3 nouveaux : test_attribution_empty_pr_body_is_manuel, test_dedup_T_Tplus5_same_issue_returns_one, test_search_issueCount_positive_control_mismatch_raises ; test_aggregate_sans_lane_aggregated_under_marker mis a jour). Retourne (services, issueCount) au lieu de list[Service] seul, pour que l'appelant puisse tracer la coherence du compteur. Refs: review c.6048082193, #18832, PR #19786 --- scripts/coordination/belt_service_balance.py | 97 +++++++++-------- scripts/tests/test_belt_service_balance.py | 106 ++++++++++++++++++- 2 files changed, 150 insertions(+), 53 deletions(-) diff --git a/scripts/coordination/belt_service_balance.py b/scripts/coordination/belt_service_balance.py index 09f978f50c..10b0ac47c7 100644 --- a/scripts/coordination/belt_service_balance.py +++ b/scripts/coordination/belt_service_balance.py @@ -187,7 +187,7 @@ def attribute_service(s: Service) -> None: Utilise le parseur canonique de `grain_tag.py`. Si pas de PR, classe `manuel`. Si PR sans tag lisible, `sans-lane` (a lister). """ - if s.pr_body is None: + if not s.pr_body: # None or empty string -> closure without PR body s.attribution_kind = "manuel" s.lane = None return @@ -495,7 +495,7 @@ def fetch_closed_issues( repo: str, window_start: datetime, max_pages: int = 30, -) -> list[Service]: +) -> tuple[list[Service], int]: """Issues CLOSED dans la fenetre, attribuees a `manuel` (sans PR). Le canal manquant du compteur historique : 278 fermetures sur 7 j ne @@ -504,28 +504,34 @@ def fetch_closed_issues( dans la fenetre et doit etre comptee comme service, attribuee a la lane `manuel` (les PRs de fermeture batch ne sont pas toutes enregistees). - Tri GraphQL sur UPDATED_AT desc (CLOSED_AT n'est pas un orderBy field) ; - on filtre en memoire sur `closedAt >= window_start`. + Tri via GraphQL `search(type: ISSUE)` avec `closed:>=` (filtre + cote serveur). Anciennement : tri UPDATED_AT desc + filtre memoire, mais + une issue fermee depuis longtemps mais commentee recemment (claim, release, + levee) remonte en tete du tri UPDATED_AT et fait tomber la fenetre trop + tot. Le filtre cote serveur evite ce piege. + + Controle positif : `issueCount` de la meme requete = nombre de + fermetures lues (apres pagination complete). Si ecart, RuntimeError. - Lecteur : `number` + `createdAt` + `closedAt`. Le `body` n'est pas - requis ici (pas d'attribution par tag, ces services vont tous a `manuel`). + Retourne : (services, issueCount). """ out: list[Service] = [] end_cursor: str | None = None pages = 0 has_next = True + expected_total = 0 # issueCount de la premiere page query = ( - "query($owner: String!, $repo: String!, $endCursor: String) {\n" - " repository(owner: $owner, name: $repo) {\n" - " issues(first: 50, after: $endCursor, " - "orderBy: {field: UPDATED_AT, direction: DESC}, " - "states: [CLOSED]) {\n" - " nodes { number createdAt closedAt }\n" - " pageInfo { hasNextPage endCursor }\n" + "query($q: String!, $endCursor: String) {\n" + " search(query: $q, type: ISSUE, first: 100, after: $endCursor) {\n" + " issueCount\n" + " nodes {\n" + " ... on Issue { number createdAt closedAt }\n" " }\n" + " pageInfo { hasNextPage endCursor }\n" " }\n" "}\n" ) + search_q = f"repo:{owner}/{repo} is:issue is:closed closed:>={window_start.strftime('%Y-%m-%d')}" token = os.environ.get("GH_TOKEN") or _run_capture(["gh", "auth", "token"]) if not token: raise RuntimeError("GH_TOKEN absent et gh auth muet (token requis)") @@ -534,8 +540,7 @@ def fetch_closed_issues( payload = json.dumps({ "query": query, "variables": { - "owner": owner, - "repo": repo, + "q": search_q, "endCursor": end_cursor, }, }) @@ -555,25 +560,13 @@ def fetch_closed_issues( raise RuntimeError(f"gh api graphql JSON parse error: {e!r}") if data.get("errors"): raise RuntimeError(f"gh api graphql errors: {data['errors']}") - nodes = ( - data.get("data", {}).get("repository", {}) - .get("issues", {}).get("nodes", []) - ) - page_earliest: datetime | None = None - for issue in nodes: + search = data.get("data", {}).get("search", {}) + if pages == 1: + expected_total = search.get("issueCount", 0) + for issue in search.get("nodes", []): closed_at = issue.get("closedAt") - if not closed_at: - continue - try: - d_close = _parse_iso(closed_at) - except Exception: - continue - if page_earliest is None or d_close < page_earliest: - page_earliest = d_close - if d_close < window_start: - continue created = issue.get("createdAt") - if not created: + if not closed_at or not created: continue svc = Service( issue_number=issue.get("number", 0), @@ -581,34 +574,40 @@ def fetch_closed_issues( service_date=closed_at, service_kind="manuel", pr_number=None, - pr_body="", + pr_body=None, # None -> attribute_service met attribution_kind = "manuel" ) attribute_service(svc) out.append(svc) - if page_earliest is not None and page_earliest < window_start - timedelta(days=14): - break - pi = ( - data.get("data", {}).get("repository", {}) - .get("issues", {}).get("pageInfo", {}) - ) + pi = search.get("pageInfo", {}) has_next = bool(pi.get("hasNextPage")) end_cursor = pi.get("endCursor") - return out + if out and len(out) != expected_total: + raise RuntimeError( + f"closed-issues count mismatch: read={len(out)} " + f"issueCount={expected_total} (search_q={search_q!r})" + ) + return out, expected_total def deduplicate_services(services: list[Service]) -> list[Service]: - """Dedoublonne les services sur (issue_number, service_date). + """Dedoublonne les services sur `issue_number` seul. + + Une issue fermee par PR est dans `pr_merge` (via closingIssuesReferences, + service_date = mergedAt) ET dans `manuel` (via closedAt, service_date = + closedAt). GitHub ferme l'issue quelques secondes apres le merge, donc + `mergedAt != closedAt` -- la cle (issue_number, service_date) ne marche + pas et chaque fermeture via PR etait comptee deux fois. Le dedup sur + `issue_number` seul resout le probleme. - Une issue fermee par PR est dans `pr_merge` (via closingIssuesReferences) - ET dans `manuel` (via closedAt). On garde le `pr_merge` (attribution par - tag `Grain:` de la PR, plus precise) et on retire le `manuel`. + On garde le `pr_merge` (attribution par tag `Grain:` de la PR, plus + precise) et on retire le `manuel`. - En cas d'egalite (deux services pour le meme issue_number + service_date - mais ni pr_merge ni manuels -- defensif), on garde le premier. + En cas d'egalite (deux services pour le meme issue_number mais ni + pr_merge ni manuels -- defensif), on garde le premier. """ - seen: dict[tuple[int, str], Service] = {} + seen: dict[int, Service] = {} for s in services: - key = (s.issue_number, s.service_date) + key = s.issue_number if key in seen: # Preferer pr_merge (attribution par tag) if seen[key].service_kind == "pr_merge": @@ -739,7 +738,7 @@ def main(argv: list[str] | None = None) -> int: else: try: services_pr = fetch_services(args.owner, args.repo, window_start, max_pages=args.max_pages) - services_manuel = fetch_closed_issues(args.owner, args.repo, window_start, max_pages=args.max_pages) + services_manuel, _closed_count = fetch_closed_issues(args.owner, args.repo, window_start, max_pages=args.max_pages) open_issues = fetch_open_issues(args.owner, args.repo, max_pages=args.max_pages) except Exception as e: print(f"UNKNOWN: {e}", file=sys.stderr) diff --git a/scripts/tests/test_belt_service_balance.py b/scripts/tests/test_belt_service_balance.py index 58b453c162..e56c64f0a7 100644 --- a/scripts/tests/test_belt_service_balance.py +++ b/scripts/tests/test_belt_service_balance.py @@ -192,17 +192,115 @@ def test_aggregate_separates_courant_tapis_ancien(): def test_aggregate_sans_lane_aggregated_under_marker(): - """Les sans-lane sont agreges sous _sans-lane, jamais absorbees ailleurs.""" + """PR sans tag lisible -> _sans-lane ; fermeture sans PR (body None ou "") + -> _manuel. Les deux categories ne se confondent jamais. + + Review c.6048082193 : avant le fix, pr_body="" (depuis fetch_closed_issues) + etait route a tort vers _sans-lane (46 fermetures dans la mesure). + Apres le fix, les deux cas (None et "") vont a _manuel, _sans-lane + reste reserve aux PRs avec body mais sans tag lisible. + """ now = datetime.now(timezone.utc) services = [ - _make_service(now - timedelta(days=2), now, pr_body=""), # sans-lane - _make_service(now - timedelta(days=3), now, pr_body=None), # manuel + _make_service(now - timedelta(days=2), now, pr_body="Description libre sans tag."), # sans-lane + _make_service(now - timedelta(days=3), now, pr_body=None), # manuel (pas de PR) + _make_service(now - timedelta(days=4), now, pr_body=""), # manuel aussi (empty body = pas de PR) ] by_lane = bsb.aggregate_by_lane(services) assert "_sans-lane" in by_lane assert "_manuel" in by_lane assert by_lane["_sans-lane"].services == 1 - assert by_lane["_manuel"].services == 1 + assert by_lane["_manuel"].services == 2 + + +def test_attribution_empty_pr_body_is_manuel(): + """pr_body="" (empty string, comme dans fetch_closed_issues) -> 'manuel', + PAS 'sans-lane'. Review c.6048082193 : le bug d'avant mettait 46 + fermetures sans PR dans `_sans-lane` au lieu de `_manuel`. + """ + s = _make_service(pr_body="") + bsb.attribute_service(s) + assert s.attribution_kind == "manuel", ( + f"pr_body='' devrait donner 'manuel', got '{s.attribution_kind}'" + ) + assert s.lane is None + + +def test_dedup_T_Tplus5_same_issue_returns_one(): + """Review c.6048082193 : mergedAt et closedAt different de quelques + secondes. Dedup sur (issue_number, service_date) laissait passer les + doublons. Apres fix (dedup sur issue_number seul), une issue fermee par + PR avec mergedAt=T et closedAt=T+5s donne 1 seul service. + """ + base = datetime(2026, 10, 5, 12, 0, 0, tzinfo=timezone.utc) + iso_t = _iso(base) + iso_t_plus_5s = _iso(base + timedelta(seconds=5)) + s_pr = bsb.Service( + issue_number=42, issue_created_at=iso_t, service_date=iso_t, + service_kind="pr_merge", pr_number=100, + pr_body="Grain: DEEP/lean -- lane myia-ai-01:CoursIA-2", + ) + s_manuel = bsb.Service( + issue_number=42, issue_created_at=iso_t, service_date=iso_t_plus_5s, + service_kind="manuel", pr_number=None, pr_body=None, + ) + bsb.attribute_service(s_pr) + bsb.attribute_service(s_manuel) + deduped = bsb.deduplicate_services([s_pr, s_manuel]) + assert len(deduped) == 1 + assert deduped[0].service_kind == "pr_merge" + assert deduped[0].lane == "myia-ai-01:CoursIA-2" + # Ordre inverse : le manuel ne doit pas ecraser le pr_merge deja vu + deduped_rev = bsb.deduplicate_services([s_manuel, s_pr]) + assert len(deduped_rev) == 1 + assert deduped_rev[0].service_kind == "pr_merge" + + +def test_search_issueCount_positive_control_mismatch_raises(): + """Si le nombre de fermetures lues != issueCount de la requete search, + fetch_closed_issues leve RuntimeError. Review c.6048082193 : le controle + positif attrape les pages manquantes ou les doublons. + + On mock subprocess.run pour rendre issueCount=10 mais seulement 3 nodes, + ce qui doit lever RuntimeError cite dans le commentaire de PR. + """ + import io + import unittest.mock as mock + + fake_response = json.dumps({ + "data": { + "search": { + "issueCount": 10, + "nodes": [ + {"number": 1, "createdAt": "2026-10-05T00:00:00Z", "closedAt": "2026-10-06T00:00:00Z"}, + {"number": 2, "createdAt": "2026-10-05T00:00:00Z", "closedAt": "2026-10-06T00:00:00Z"}, + {"number": 3, "createdAt": "2026-10-05T00:00:00Z", "closedAt": "2026-10-06T00:00:00Z"}, + ], + "pageInfo": {"hasNextPage": False, "endCursor": None}, + } + } + }) + fake_completed = mock.Mock(returncode=0, stdout=fake_response, stderr="") + + with mock.patch("subprocess.run", return_value=fake_completed), \ + mock.patch.dict("os.environ", {"GH_TOKEN": "fake-token"}, clear=False), \ + mock.patch("belt_service_balance._run_capture", return_value="fake-token"), \ + mock.patch("sys.stderr", io.StringIO()): + try: + bsb.fetch_closed_issues( + "jsboige", "CoursIA", + datetime(2026, 10, 1, tzinfo=timezone.utc), + max_pages=1, + ) + raised = False + msg = "" + except RuntimeError as e: + raised = True + msg = str(e) + assert raised, "fetch_closed_issues aurait du lever RuntimeError (read=3 vs issueCount=10)" + assert "closed-issues count mismatch" in msg + assert "read=3" in msg + assert "issueCount=10" in msg # ---------------------------------------------------------------------------