From 48903acaa5b6769229d2144f74bf90e83c470de4 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 26 Jun 2026 14:40:05 +0000 Subject: [PATCH] Implement Intel CPU AI triage toolkit (all 7 phases) Self-contained, stdlib-only toolkit under tools/intel_cpu_triage/ implementing the AI-assisted Intel CPU issue resolution workflow: incremental GitHub ingestion, AI/heuristic triage and scoring into three buckets, duplicate detection, a board state machine with two human gates, PR rate-limiting, and health metrics with a feedback loop. Includes CLI, example scheduled workflow, README, and 26 passing unit tests. Authored with assistance from Claude. --- tools/intel_cpu_triage/README.md | 116 ++++++ tools/intel_cpu_triage/__init__.py | 26 ++ tools/intel_cpu_triage/board.py | 75 ++++ tools/intel_cpu_triage/cli.py | 194 +++++++++ tools/intel_cpu_triage/config.example.json | 47 +++ tools/intel_cpu_triage/config.py | 135 +++++++ tools/intel_cpu_triage/db.py | 372 ++++++++++++++++++ tools/intel_cpu_triage/dedup.py | 102 +++++ tools/intel_cpu_triage/github_client.py | 99 +++++ tools/intel_cpu_triage/ingest.py | 101 +++++ .../intel_cpu_triage.yml.example | 53 +++ tools/intel_cpu_triage/metrics.py | 101 +++++ tools/intel_cpu_triage/ratelimit.py | 54 +++ tools/intel_cpu_triage/tests/__init__.py | 5 + tools/intel_cpu_triage/tests/test_toolkit.py | 332 ++++++++++++++++ tools/intel_cpu_triage/triage.py | 299 ++++++++++++++ 16 files changed, 2111 insertions(+) create mode 100644 tools/intel_cpu_triage/README.md create mode 100644 tools/intel_cpu_triage/__init__.py create mode 100644 tools/intel_cpu_triage/board.py create mode 100644 tools/intel_cpu_triage/cli.py create mode 100644 tools/intel_cpu_triage/config.example.json create mode 100644 tools/intel_cpu_triage/config.py create mode 100644 tools/intel_cpu_triage/db.py create mode 100644 tools/intel_cpu_triage/dedup.py create mode 100644 tools/intel_cpu_triage/github_client.py create mode 100644 tools/intel_cpu_triage/ingest.py create mode 100644 tools/intel_cpu_triage/intel_cpu_triage.yml.example create mode 100644 tools/intel_cpu_triage/metrics.py create mode 100644 tools/intel_cpu_triage/ratelimit.py create mode 100644 tools/intel_cpu_triage/tests/__init__.py create mode 100644 tools/intel_cpu_triage/tests/test_toolkit.py create mode 100644 tools/intel_cpu_triage/triage.py diff --git a/tools/intel_cpu_triage/README.md b/tools/intel_cpu_triage/README.md new file mode 100644 index 0000000000000..b49167f6f09f4 --- /dev/null +++ b/tools/intel_cpu_triage/README.md @@ -0,0 +1,116 @@ +# Intel CPU Issue Triage Toolkit + +An AI-assisted workflow for a small (3–5 engineer) team resolving Intel +CPU / oneDNN issues in [`pytorch/pytorch`](https://github.com/pytorch/pytorch), +**without** flooding maintainers with low-quality PRs. + +This toolkit implements the plan as runnable code. It is **standard-library +only** (no third-party dependencies, no PyTorch build required) so it can run +anywhere, including a scheduled GitHub Action. + +## Guiding principles + +1. **Reviewer trust is the scarcest resource.** Optimize for PR *acceptance + rate* and reviewer goodwill, not PR count. +2. **AI is an accelerator, not an autonomous submitter.** A human engineer owns + every issue and every PR. AI does triage, reproduction, drafting, and + tracking; humans decide what ships. + +Two **mandatory human gates** are baked into the flow: +- **Gate #1 (grooming):** an engineer must `pull` an issue onto the board. +- **Gate #2 (internal review):** a second engineer signs off before any PR is + opened upstream, subject to the rate-limit policy. + +## Phases → modules + +| Phase | Purpose | Module | +|------:|---------|--------| +| 1 | Ingest relevant issues into a local SQLite DB (incremental, watermark-based) | `ingest.py`, `github_client.py` | +| 2 | AI triage & scoring into 3 buckets + duplicate detection | `triage.py`, `dedup.py` | +| 3 | Tracking board state machine (Kanban) | `board.py` | +| 5/6 | Rate-limiting (open-PR cap + per-reviewer budget) | `ratelimit.py` | +| 7 | Health metrics + feedback examples for prompt tuning | `metrics.py` | +| — | Single source of truth | `db.py` | +| — | Orchestration | `cli.py` | + +> The DB is the single source of truth. Upstream GitHub labels are never +> modified by this toolkit. + +## The three triage buckets (Phase 2) + +- **`ready_to_work`** — in-scope, has a repro, scores above threshold. +- **`needs_info_repro`** — missing a reproduction or below threshold. +- **`needs_maintainer_decision`** — *quarantined*: already has an open PR, is a + feature request / design discussion, or implies an API/semantics change that + needs an RFC. + +Quarantining is what keeps the team from chasing work that would generate +unwelcome PRs. + +## Usage + +```bash +# Phase 1 — incremental ingestion (needs a GITHUB_TOKEN env var for higher rate limits) +python -m tools.intel_cpu_triage.cli --config tools/intel_cpu_triage/config.example.json ingest + +# Phase 2 — score & bucket (heuristic by default; see "Wiring an LLM" below) +python -m tools.intel_cpu_triage.cli triage + +# Phase 3 — review the AI-proposed list, then a human pulls work onto the board +python -m tools.intel_cpu_triage.cli ready +python -m tools.intel_cpu_triage.cli pull 12345 alice # human gate #1 +python -m tools.intel_cpu_triage.cli advance 12345 repro_confirmed +python -m tools.intel_cpu_triage.cli board + +# Phase 6 — before opening an upstream PR, check the policy (human gate #2) +python -m tools.intel_cpu_triage.cli pr-check --reviewer some-maintainer + +# Phase 7 — health metrics and feedback examples +python -m tools.intel_cpu_triage.cli metrics +python -m tools.intel_cpu_triage.cli feedback +``` + +The board enforces legal transitions +(`backlog → investigating → repro_confirmed → fix_in_progress → pr_open → +merged`); you cannot skip repro confirmation to jump straight to a PR. + +## Wiring an LLM (optional) + +`triage` uses the deterministic `HeuristicTriager` by default so it works +offline. To use a real model, write a tiny driver that constructs +`triage.LLMTriager(cfg, llm)` where `llm` has a `complete(prompt) -> str` +method returning JSON. The toolkit: + +- merges the model's fields onto the heuristic baseline (so a partial or garbled + response still yields a complete, valid record), and +- derives the bucket itself from the merged fields, keeping bucketing policy in + one place. + +Feed `feedback` output back into your prompt as few-shot "good vs. bad +candidate" examples to improve triage precision over time. + +## Scheduling + +Copy `intel_cpu_triage.yml.example` into your **private** tracking repo at +`.github/workflows/intel_cpu_triage.yml`. It runs ingestion + triage daily and +caches the SQLite DB between runs to keep ingestion incremental. (Shipped with a +`.example` suffix so it never auto-runs from a PyTorch fork.) + +## Configuration + +See `config.example.json`. Key knobs: scope/signal/negative labels, content +keywords, `max_open_external_prs`, `max_prs_per_reviewer`, and +`ready_score_threshold`. + +## Tests + +```bash +python -m unittest tools.intel_cpu_triage.tests.test_toolkit +``` + +## What this toolkit deliberately does NOT do + +- It never auto-posts comments or auto-opens PRs on `pytorch/pytorch`. +- It never modifies upstream labels. +- It does not advance API-change / feature work without a maintainer decision. +- It does not optimize for PR volume. diff --git a/tools/intel_cpu_triage/__init__.py b/tools/intel_cpu_triage/__init__.py new file mode 100644 index 0000000000000..68aed04a03d64 --- /dev/null +++ b/tools/intel_cpu_triage/__init__.py @@ -0,0 +1,26 @@ +"""AI-assisted Intel CPU issue resolution toolkit for pytorch/pytorch. + +This package implements the workflow described in ``README.md``: incremental +ingestion of relevant issues from GitHub, AI-assisted triage and scoring, +duplicate detection, a tracking board state machine, rate-limiting to protect +reviewer trust, and health metrics. + +The toolkit is intentionally dependency-free (standard library only) so it can +run in restricted environments and CI without a PyTorch build. +""" + +from __future__ import annotations + +__all__ = [ + "config", + "db", + "github_client", + "ingest", + "triage", + "dedup", + "board", + "ratelimit", + "metrics", +] + +__version__ = "0.1.0" diff --git a/tools/intel_cpu_triage/board.py b/tools/intel_cpu_triage/board.py new file mode 100644 index 0000000000000..2d6dd14e82a74 --- /dev/null +++ b/tools/intel_cpu_triage/board.py @@ -0,0 +1,75 @@ +"""Phase 3: tracking board state machine. + +The board mirrors engineering progress for issues a human has *pulled* into the +sprint. AI proposes the ready-to-work list (Phase 2); only an explicit +``pull_to_board`` call (a human action in practice) moves work forward. This +keeps a human accountable for every issue picked up. +""" + +from __future__ import annotations + +from .db import BOARD_COLUMNS, BUCKET_READY, Database + + +# Allowed forward/backward transitions. Backwards moves (e.g. back to +# investigating) are permitted because real work is non-linear, but skipping +# ahead past repro confirmation into a PR is not. +_ALLOWED = { + "backlog": {"investigating", "closed"}, + "investigating": {"repro_confirmed", "needs_info", "backlog", "closed"}, + "repro_confirmed": {"fix_in_progress", "investigating", "closed"}, + "fix_in_progress": {"pr_open", "repro_confirmed", "closed"}, + "pr_open": {"merged", "fix_in_progress", "closed"}, + "merged": set(), + "closed": set(), +} + + +class BoardError(ValueError): + pass + + +def ready_list(db: Database, limit: int | None = None) -> list: + """Return AI-proposed ready-to-work issues, highest actionability first.""" + issues = db.issues_in_bucket(BUCKET_READY) + return issues[:limit] if limit else issues + + +def pull_to_board(db: Database, number: int, assignee: str) -> None: + """Human gate #1: assign an engineer and start investigation. + + Only issues that triage marked ready-to-work may be pulled. This is the + enforcement point for "AI proposes, humans dispose". + """ + issue = db.get_issue(number) + if issue is None: + raise BoardError(f"unknown issue #{number}") + if issue.bucket != BUCKET_READY: + raise BoardError( + f"issue #{number} is in bucket {issue.bucket!r}, not ready-to-work; " + "it must be re-triaged or handled via the appropriate gate" + ) + db.set_board(number, "investigating", assignee=assignee) + + +def advance(db: Database, number: int, to_column: str) -> None: + """Move an issue to a new board column, validating the transition.""" + if to_column not in BOARD_COLUMNS and to_column != "needs_info": + raise BoardError(f"unknown column {to_column!r}") + issue = db.get_issue(number) + if issue is None: + raise BoardError(f"unknown issue #{number}") + current = issue.board_column + allowed = _ALLOWED.get(current, set()) + if to_column not in allowed: + raise BoardError( + f"illegal transition for #{number}: {current} -> {to_column}; " + f"allowed: {sorted(allowed)}" + ) + target = "backlog" if to_column == "needs_info" else to_column + db.set_board(number, target) + + +def board_snapshot(db: Database) -> dict[str, list[int]]: + """Return issue numbers grouped by board column (for display/export).""" + return {col: [i.number for i in db.issues_in_column(col)] for col in BOARD_COLUMNS} diff --git a/tools/intel_cpu_triage/cli.py b/tools/intel_cpu_triage/cli.py new file mode 100644 index 0000000000000..d99ca71e33b87 --- /dev/null +++ b/tools/intel_cpu_triage/cli.py @@ -0,0 +1,194 @@ +"""Command-line interface for the Intel CPU triage toolkit. + +Subcommands map onto workflow phases: + + ingest Phase 1 pull relevant issues into the local DB + triage Phase 2 score & bucket issues (+ duplicate detection) + ready Phase 3 show the AI-proposed ready-to-work list + pull Phase 3 assign an engineer (human gate #1) + advance Phase 3 move an issue along the board + board Phase 3 print the board snapshot + pr-check Phase 6 check whether a new upstream PR is within policy + metrics Phase 7 print health metrics + feedback Phase 7 export accepted/rejected examples for prompt tuning + +The triage subcommand uses the heuristic triager by default. Wire an LLM by +importing :class:`triage.LLMTriager` from your own driver script. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from typing import Sequence + +from . import board, metrics, ratelimit +from .config import load_config +from .db import Database +from .dedup import SimilarityIndex +from .github_client import GitHubClient +from .ingest import ingest +from .triage import HeuristicTriager + + +def _db(args: argparse.Namespace): + cfg = load_config(args.config) + if args.db: + cfg.db_path = args.db + return Database(cfg.db_path), cfg + + +def cmd_ingest(args: argparse.Namespace) -> int: + db, cfg = _db(args) + token = os.environ.get("GITHUB_TOKEN") + client = GitHubClient(token=token) + changed = ingest(db, client, cfg) + print(f"ingested/updated {len(changed)} issues") + return 0 + + +def cmd_triage(args: argparse.Namespace) -> int: + db, cfg = _db(args) + issues = db.all_issues() + index = SimilarityIndex() + index.build(issues) + triager = HeuristicTriager(cfg) + for issue in issues: + if not args.all and issue.bucket is not None: + continue + related = [n for n, _ in index.most_similar(issue, top_k=3, threshold=0.2)] + result = triager.triage(issue, related=related) + db.save_triage( + issue.number, + result.bucket, + result.actionability, + result.confidence, + result.to_dict(), + ) + counts: dict[str, int] = {} + for issue in db.all_issues(): + if issue.bucket: + counts[issue.bucket] = counts.get(issue.bucket, 0) + 1 + print(json.dumps(counts, indent=2)) + return 0 + + +def cmd_ready(args: argparse.Namespace) -> int: + db, _ = _db(args) + for issue in board.ready_list(db, limit=args.limit): + subsystems = "" + if issue.triage: + subsystems = ", ".join(issue.triage.get("suspected_subsystems", [])) + print( + f"#{issue.number:<7} score={issue.actionability:<3} " + f"conf={issue.confidence} {issue.title[:70]}" + + (f" [{subsystems}]" if subsystems else "") + ) + return 0 + + +def cmd_pull(args: argparse.Namespace) -> int: + db, _ = _db(args) + try: + board.pull_to_board(db, args.number, args.assignee) + except board.BoardError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + print(f"#{args.number} -> investigating (assignee: {args.assignee})") + return 0 + + +def cmd_advance(args: argparse.Namespace) -> int: + db, _ = _db(args) + try: + board.advance(db, args.number, args.to) + except board.BoardError as exc: + print(f"error: {exc}", file=sys.stderr) + return 1 + print(f"#{args.number} -> {args.to}") + return 0 + + +def cmd_board(args: argparse.Namespace) -> int: + db, _ = _db(args) + print(json.dumps(board.board_snapshot(db), indent=2)) + return 0 + + +def cmd_pr_check(args: argparse.Namespace) -> int: + db, cfg = _db(args) + decision = ratelimit.can_open_pr(db, cfg, reviewer=args.reviewer) + if decision: + print("OK: opening a PR is within policy") + return 0 + print("BLOCKED: " + "; ".join(decision.reasons), file=sys.stderr) + return 2 + + +def cmd_metrics(args: argparse.Namespace) -> int: + db, _ = _db(args) + print(json.dumps(metrics.compute_metrics(db).to_dict(), indent=2)) + return 0 + + +def cmd_feedback(args: argparse.Namespace) -> int: + db, _ = _db(args) + print(json.dumps(metrics.feedback_examples(db), indent=2)) + return 0 + + +def build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser(prog="intel_cpu_triage", description=__doc__) + p.add_argument("--config", help="path to config JSON", default=None) + p.add_argument("--db", help="path to SQLite DB (overrides config)", default=None) + sub = p.add_subparsers(dest="command", required=True) + + sub.add_parser("ingest", help="Phase 1: pull relevant issues").set_defaults( + func=cmd_ingest + ) + + t = sub.add_parser("triage", help="Phase 2: score & bucket issues") + t.add_argument("--all", action="store_true", help="re-triage already-triaged issues") + t.set_defaults(func=cmd_triage) + + r = sub.add_parser("ready", help="Phase 3: list ready-to-work issues") + r.add_argument("--limit", type=int, default=20) + r.set_defaults(func=cmd_ready) + + pl = sub.add_parser("pull", help="Phase 3: assign an engineer (human gate #1)") + pl.add_argument("number", type=int) + pl.add_argument("assignee") + pl.set_defaults(func=cmd_pull) + + av = sub.add_parser("advance", help="Phase 3: move an issue along the board") + av.add_argument("number", type=int) + av.add_argument("to", help="target column") + av.set_defaults(func=cmd_advance) + + sub.add_parser("board", help="Phase 3: print board snapshot").set_defaults( + func=cmd_board + ) + + pc = sub.add_parser("pr-check", help="Phase 6: is a new upstream PR allowed?") + pc.add_argument("--reviewer", default=None) + pc.set_defaults(func=cmd_pr_check) + + sub.add_parser("metrics", help="Phase 7: health metrics").set_defaults( + func=cmd_metrics + ) + sub.add_parser("feedback", help="Phase 7: export scoring feedback").set_defaults( + func=cmd_feedback + ) + return p + + +def main(argv: Sequence[str] | None = None) -> int: + parser = build_parser() + args = parser.parse_args(argv) + return args.func(args) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/intel_cpu_triage/config.example.json b/tools/intel_cpu_triage/config.example.json new file mode 100644 index 0000000000000..4aed6818b390a --- /dev/null +++ b/tools/intel_cpu_triage/config.example.json @@ -0,0 +1,47 @@ +{ + "repo": "pytorch/pytorch", + "scope_labels": [ + "module: cpu", + "module: mkldnn", + "module: amp", + "module: half", + "module: performance", + "oncall: quantization" + ], + "signal_labels": [ + "triaged", + "triage review", + "high priority", + "small", + "good first issue" + ], + "negative_labels": [ + "needs reproduction", + "module: dependency bug", + "feature", + "enhancement", + "proposal accepted" + ], + "content_keywords": [ + "amx", + "onednn", + "mkldnn", + "ideep", + "avx512", + "avx2", + "vnni", + "bf16", + "bfloat16", + "fp16", + "xeon", + "sapphire rapids", + "granite rapids", + "intel", + "cpublas", + "brgemm" + ], + "max_open_external_prs": 5, + "max_prs_per_reviewer": 2, + "ready_score_threshold": 60, + "db_path": "intel_cpu_triage.db" +} diff --git a/tools/intel_cpu_triage/config.py b/tools/intel_cpu_triage/config.py new file mode 100644 index 0000000000000..b0d6f69fb0314 --- /dev/null +++ b/tools/intel_cpu_triage/config.py @@ -0,0 +1,135 @@ +"""Configuration for the Intel CPU triage toolkit. + +Defaults encode the labels and content keywords that are relevant to an Intel +CPU / oneDNN team working on ``pytorch/pytorch``. They can be overridden by a +JSON file (see ``config.example.json``) loaded via :func:`load_config`. +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field, asdict +from pathlib import Path + + +# Labels that, on their own, make an issue in-scope for an Intel CPU team. +DEFAULT_SCOPE_LABELS: list[str] = [ + "module: cpu", + "module: mkldnn", + "module: amp", + "module: half", + "module: performance", + "oncall: quantization", +] + +# Labels that maintainers use to signal an issue is worth working on. Presence +# of these raises an issue's priority during scoring. +DEFAULT_SIGNAL_LABELS: list[str] = [ + "triaged", + "triage review", + "high priority", + "small", + "good first issue", +] + +# Labels that should lower priority or quarantine an issue. +DEFAULT_NEGATIVE_LABELS: list[str] = [ + "needs reproduction", + "module: dependency bug", + "feature", + "enhancement", + "proposal accepted", +] + +# Content keywords implying Intel hardware even when an issue is mislabeled. +# Matched case-insensitively against title + body. +DEFAULT_CONTENT_KEYWORDS: list[str] = [ + "amx", + "onednn", + "mkldnn", + "mkl-dnn", + "ideep", + "avx512", + "avx-512", + "avx2", + "vnni", + "bf16", + "bfloat16", + "fp16", + "xeon", + "sapphire rapids", + "granite rapids", + "emerald rapids", + "intel", + "cpublas", + "brgemm", +] + +# Subsystem hints: keyword -> suspected source path. Used by the heuristic +# triager to populate "suspected files/subsystem". +DEFAULT_SUBSYSTEM_HINTS: dict[str, str] = { + "mkldnn": "aten/src/ATen/native/mkldnn/", + "onednn": "aten/src/ATen/native/mkldnn/", + "ideep": "third_party/ideep/", + "brgemm": "aten/src/ATen/native/CPUBlas.cpp", + "cpublas": "aten/src/ATen/native/CPUBlas.cpp", + "quant": "aten/src/ATen/native/quantized/cpu/", + "avx512": "aten/src/ATen/cpu/vec/", + "avx2": "aten/src/ATen/cpu/vec/", + "vec256": "aten/src/ATen/cpu/vec/", + "bf16": "aten/src/ATen/cpu/vec/", + "openmp": "aten/src/ATen/ParallelOpenMP.h", +} + + +@dataclass +class Config: + """Toolkit configuration.""" + + repo: str = "pytorch/pytorch" + scope_labels: list[str] = field(default_factory=lambda: list(DEFAULT_SCOPE_LABELS)) + signal_labels: list[str] = field( + default_factory=lambda: list(DEFAULT_SIGNAL_LABELS) + ) + negative_labels: list[str] = field( + default_factory=lambda: list(DEFAULT_NEGATIVE_LABELS) + ) + content_keywords: list[str] = field( + default_factory=lambda: list(DEFAULT_CONTENT_KEYWORDS) + ) + subsystem_hints: dict[str, str] = field( + default_factory=lambda: dict(DEFAULT_SUBSYSTEM_HINTS) + ) + + # Phase 6 rate-limiting policy. + max_open_external_prs: int = 5 + max_prs_per_reviewer: int = 2 + + # Phase 2 scoring: minimum actionability score (0-100) to land in + # "ready-to-work". + ready_score_threshold: int = 60 + + db_path: str = "intel_cpu_triage.db" + + def to_json(self) -> str: + return json.dumps(asdict(self), indent=2, sort_keys=True) + + +def load_config(path: str | Path | None = None) -> Config: + """Load configuration from a JSON file, falling back to defaults. + + Unknown keys in the file are ignored so that newer config files remain + loadable by older code. + """ + cfg = Config() + if path is None: + return cfg + p = Path(path) + if not p.exists(): + return cfg + data = json.loads(p.read_text()) + known = set(asdict(cfg).keys()) + for key, value in data.items(): + if key in known: + setattr(cfg, key, value) + return cfg diff --git a/tools/intel_cpu_triage/db.py b/tools/intel_cpu_triage/db.py new file mode 100644 index 0000000000000..76bd1e39a8ead --- /dev/null +++ b/tools/intel_cpu_triage/db.py @@ -0,0 +1,372 @@ +"""SQLite persistence: the toolkit's single source of truth. + +The database keeps issue records, triage results, board state, and tracked +pull requests. It is intentionally a thin layer over :mod:`sqlite3` so it has +no third-party dependencies and is trivial to inspect with the ``sqlite3`` CLI. +""" + +from __future__ import annotations + +import json +import sqlite3 +import time +from contextlib import contextmanager +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Iterator + + +# Kanban columns (Phase 3). ``BACKLOG`` is the entry point for triaged, +# ready-to-work issues; the rest mirror engineering progress. +BOARD_COLUMNS: tuple[str, ...] = ( + "backlog", + "investigating", + "repro_confirmed", + "fix_in_progress", + "pr_open", + "merged", + "closed", +) + +# Triage buckets (Phase 2). +BUCKET_READY = "ready_to_work" +BUCKET_NEEDS_INFO = "needs_info_repro" +BUCKET_NEEDS_DECISION = "needs_maintainer_decision" +BUCKETS: tuple[str, ...] = (BUCKET_READY, BUCKET_NEEDS_INFO, BUCKET_NEEDS_DECISION) + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS issues ( + number INTEGER PRIMARY KEY, + title TEXT NOT NULL, + body TEXT, + state TEXT, + author TEXT, + author_association TEXT, + labels TEXT, -- JSON array + reactions INTEGER DEFAULT 0, + comments INTEGER DEFAULT 0, + linked_prs TEXT, -- JSON array of PR numbers + created_at TEXT, + updated_at TEXT, + fetched_at REAL, + -- triage (Phase 2) + bucket TEXT, + actionability INTEGER, + confidence REAL, + triage TEXT, -- JSON blob of full triage record + triaged_at REAL, + -- board (Phase 3) + board_column TEXT DEFAULT 'backlog', + assignee TEXT, + board_updated_at REAL +); + +CREATE TABLE IF NOT EXISTS pull_requests ( + number INTEGER PRIMARY KEY, + issue_number INTEGER, + reviewer TEXT, + state TEXT, -- open | merged | closed + changes_requested INTEGER DEFAULT 0, + opened_at TEXT, + closed_at TEXT, + merged_at TEXT, + url TEXT +); + +CREATE TABLE IF NOT EXISTS meta ( + key TEXT PRIMARY KEY, + value TEXT +); + +CREATE INDEX IF NOT EXISTS idx_issues_bucket ON issues(bucket); +CREATE INDEX IF NOT EXISTS idx_issues_board ON issues(board_column); +CREATE INDEX IF NOT EXISTS idx_pr_state ON pull_requests(state); +""" + + +@dataclass +class Issue: + number: int + title: str + body: str = "" + state: str = "open" + author: str = "" + author_association: str = "" + labels: list[str] = field(default_factory=list) + reactions: int = 0 + comments: int = 0 + linked_prs: list[int] = field(default_factory=list) + created_at: str = "" + updated_at: str = "" + fetched_at: float = 0.0 + bucket: str | None = None + actionability: int | None = None + confidence: float | None = None + triage: dict[str, Any] | None = None + triaged_at: float | None = None + board_column: str = "backlog" + assignee: str | None = None + board_updated_at: float | None = None + + @property + def text(self) -> str: + """Combined title + body for keyword/semantic matching.""" + return f"{self.title}\n\n{self.body or ''}" + + +def _row_to_issue(row: sqlite3.Row) -> Issue: + return Issue( + number=row["number"], + title=row["title"], + body=row["body"] or "", + state=row["state"] or "open", + author=row["author"] or "", + author_association=row["author_association"] or "", + labels=json.loads(row["labels"]) if row["labels"] else [], + reactions=row["reactions"] or 0, + comments=row["comments"] or 0, + linked_prs=json.loads(row["linked_prs"]) if row["linked_prs"] else [], + created_at=row["created_at"] or "", + updated_at=row["updated_at"] or "", + fetched_at=row["fetched_at"] or 0.0, + bucket=row["bucket"], + actionability=row["actionability"], + confidence=row["confidence"], + triage=json.loads(row["triage"]) if row["triage"] else None, + triaged_at=row["triaged_at"], + board_column=row["board_column"] or "backlog", + assignee=row["assignee"], + board_updated_at=row["board_updated_at"], + ) + + +class Database: + """Thin SQLite wrapper for the triage toolkit.""" + + def __init__(self, path: str | Path = ":memory:") -> None: + self.path = str(path) + self.conn = sqlite3.connect(self.path) + self.conn.row_factory = sqlite3.Row + self.conn.execute("PRAGMA foreign_keys = ON") + self.conn.executescript(SCHEMA) + self.conn.commit() + + def close(self) -> None: + self.conn.close() + + @contextmanager + def _tx(self) -> Iterator[sqlite3.Connection]: + try: + yield self.conn + self.conn.commit() + except Exception: + self.conn.rollback() + raise + + # -- issues ----------------------------------------------------------- + def upsert_issue(self, issue: Issue) -> None: + """Insert or update an issue, preserving triage/board fields on update. + + Ingestion (Phase 1) only owns the raw GitHub fields; it must not clobber + triage or board state set by later phases. We therefore merge: ingest + fields are overwritten, but triage/board columns are kept unless this + record explicitly carries new values. + """ + existing = self.get_issue(issue.number) + if existing is not None: + # Preserve downstream state unless the incoming issue sets it. + if issue.bucket is None: + issue.bucket = existing.bucket + issue.actionability = existing.actionability + issue.confidence = existing.confidence + issue.triage = existing.triage + issue.triaged_at = existing.triaged_at + if issue.board_updated_at is None: + issue.board_column = existing.board_column + issue.assignee = existing.assignee + issue.board_updated_at = existing.board_updated_at + with self._tx() as conn: + conn.execute( + """ + INSERT INTO issues ( + number, title, body, state, author, author_association, + labels, reactions, comments, linked_prs, created_at, + updated_at, fetched_at, bucket, actionability, confidence, + triage, triaged_at, board_column, assignee, board_updated_at + ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(number) DO UPDATE SET + title=excluded.title, + body=excluded.body, + state=excluded.state, + author=excluded.author, + author_association=excluded.author_association, + labels=excluded.labels, + reactions=excluded.reactions, + comments=excluded.comments, + linked_prs=excluded.linked_prs, + created_at=excluded.created_at, + updated_at=excluded.updated_at, + fetched_at=excluded.fetched_at, + bucket=excluded.bucket, + actionability=excluded.actionability, + confidence=excluded.confidence, + triage=excluded.triage, + triaged_at=excluded.triaged_at, + board_column=excluded.board_column, + assignee=excluded.assignee, + board_updated_at=excluded.board_updated_at + """, + ( + issue.number, + issue.title, + issue.body, + issue.state, + issue.author, + issue.author_association, + json.dumps(issue.labels), + issue.reactions, + issue.comments, + json.dumps(issue.linked_prs), + issue.created_at, + issue.updated_at, + issue.fetched_at, + issue.bucket, + issue.actionability, + issue.confidence, + json.dumps(issue.triage) if issue.triage is not None else None, + issue.triaged_at, + issue.board_column, + issue.assignee, + issue.board_updated_at, + ), + ) + + def get_issue(self, number: int) -> Issue | None: + row = self.conn.execute( + "SELECT * FROM issues WHERE number = ?", (number,) + ).fetchone() + return _row_to_issue(row) if row else None + + def all_issues(self) -> list[Issue]: + rows = self.conn.execute( + "SELECT * FROM issues ORDER BY number" + ).fetchall() + return [_row_to_issue(r) for r in rows] + + def issues_in_bucket(self, bucket: str) -> list[Issue]: + rows = self.conn.execute( + "SELECT * FROM issues WHERE bucket = ? ORDER BY " + "actionability DESC, reactions DESC", + (bucket,), + ).fetchall() + return [_row_to_issue(r) for r in rows] + + def issues_in_column(self, column: str) -> list[Issue]: + rows = self.conn.execute( + "SELECT * FROM issues WHERE board_column = ? ORDER BY number", + (column,), + ).fetchall() + return [_row_to_issue(r) for r in rows] + + def save_triage( + self, + number: int, + bucket: str, + actionability: int, + confidence: float, + triage: dict[str, Any], + ) -> None: + with self._tx() as conn: + conn.execute( + "UPDATE issues SET bucket=?, actionability=?, confidence=?, " + "triage=?, triaged_at=? WHERE number=?", + ( + bucket, + actionability, + confidence, + json.dumps(triage), + time.time(), + number, + ), + ) + + def set_board( + self, number: int, column: str, assignee: str | None = None + ) -> None: + if column not in BOARD_COLUMNS: + raise ValueError(f"unknown board column: {column}") + with self._tx() as conn: + conn.execute( + "UPDATE issues SET board_column=?, assignee=COALESCE(?, assignee), " + "board_updated_at=? WHERE number=?", + (column, assignee, time.time(), number), + ) + + # -- pull requests ---------------------------------------------------- + def upsert_pr( + self, + number: int, + issue_number: int | None = None, + reviewer: str | None = None, + state: str = "open", + changes_requested: int = 0, + opened_at: str = "", + closed_at: str = "", + merged_at: str = "", + url: str = "", + ) -> None: + with self._tx() as conn: + conn.execute( + """ + INSERT INTO pull_requests ( + number, issue_number, reviewer, state, changes_requested, + opened_at, closed_at, merged_at, url + ) VALUES (?,?,?,?,?,?,?,?,?) + ON CONFLICT(number) DO UPDATE SET + issue_number=COALESCE(excluded.issue_number, issue_number), + reviewer=COALESCE(excluded.reviewer, reviewer), + state=excluded.state, + changes_requested=excluded.changes_requested, + closed_at=excluded.closed_at, + merged_at=excluded.merged_at, + url=excluded.url + """, + ( + number, + issue_number, + reviewer, + state, + changes_requested, + opened_at, + closed_at, + merged_at, + url, + ), + ) + + def all_prs(self) -> list[dict[str, Any]]: + rows = self.conn.execute( + "SELECT * FROM pull_requests ORDER BY number" + ).fetchall() + return [dict(r) for r in rows] + + def open_prs(self) -> list[dict[str, Any]]: + rows = self.conn.execute( + "SELECT * FROM pull_requests WHERE state = 'open' ORDER BY number" + ).fetchall() + return [dict(r) for r in rows] + + # -- meta / watermark ------------------------------------------------- + def get_meta(self, key: str, default: str | None = None) -> str | None: + row = self.conn.execute( + "SELECT value FROM meta WHERE key = ?", (key,) + ).fetchone() + return row["value"] if row else default + + def set_meta(self, key: str, value: str) -> None: + with self._tx() as conn: + conn.execute( + "INSERT INTO meta (key, value) VALUES (?, ?) " + "ON CONFLICT(key) DO UPDATE SET value=excluded.value", + (key, value), + ) diff --git a/tools/intel_cpu_triage/dedup.py b/tools/intel_cpu_triage/dedup.py new file mode 100644 index 0000000000000..d1c282a5dabee --- /dev/null +++ b/tools/intel_cpu_triage/dedup.py @@ -0,0 +1,102 @@ +"""Duplicate / related detection (part of Phase 2). + +Uses a dependency-free TF-IDF cosine similarity over issue text. An optional +``embedder`` callable can be supplied to use real embeddings; if it raises or +is absent, the toolkit falls back to the lexical model so it always works +offline. +""" + +from __future__ import annotations + +import math +import re +from collections import Counter +from typing import Callable, Iterable + +from .db import Issue + + +_TOKEN_RE = re.compile(r"[a-z0-9_]+") +# Very common English/markdown tokens that add noise to similarity. +_STOPWORDS = frozenset( + """ + the a an and or but if then for to of in on at by with without is are was + be been being this that these those it its as from into out up down i we + you he she they them me my our your when how what why which who whom can + could should would will shall do does did not no yes so such than too very + bug issue error code python torch import return def self none true false + """.split() +) + + +def tokenize(text: str) -> list[str]: + return [t for t in _TOKEN_RE.findall(text.lower()) if t not in _STOPWORDS] + + +def _tf(tokens: list[str]) -> dict[str, float]: + if not tokens: + return {} + counts = Counter(tokens) + n = len(tokens) + return {term: c / n for term, c in counts.items()} + + +def _cosine(a: dict[str, float], b: dict[str, float]) -> float: + if not a or not b: + return 0.0 + common = set(a) & set(b) + dot = sum(a[t] * b[t] for t in common) + na = math.sqrt(sum(v * v for v in a.values())) + nb = math.sqrt(sum(v * v for v in b.values())) + if na == 0 or nb == 0: + return 0.0 + return dot / (na * nb) + + +class SimilarityIndex: + """Lexical TF-IDF index over a corpus of issues. + + ``most_similar`` returns issues ranked by cosine similarity, excluding the + query issue itself. + """ + + def __init__(self, embedder: Callable[[str], list[float]] | None = None) -> None: + self._embedder = embedder + self._docs: dict[int, dict[str, float]] = {} + self._df: Counter[str] = Counter() + self._n = 0 + + def build(self, issues: Iterable[Issue]) -> None: + issues = list(issues) + self._n = len(issues) + self._df = Counter() + token_lists: dict[int, list[str]] = {} + for issue in issues: + tokens = tokenize(issue.text) + token_lists[issue.number] = tokens + for term in set(tokens): + self._df[term] += 1 + self._docs = { + num: self._tfidf(tokens) for num, tokens in token_lists.items() + } + + def _idf(self, term: str) -> float: + df = self._df.get(term, 0) + return math.log((1 + self._n) / (1 + df)) + 1.0 + + def _tfidf(self, tokens: list[str]) -> dict[str, float]: + return {term: tf * self._idf(term) for term, tf in _tf(tokens).items()} + + def most_similar( + self, issue: Issue, top_k: int = 5, threshold: float = 0.0 + ) -> list[tuple[int, float]]: + query = self._tfidf(tokenize(issue.text)) + scored: list[tuple[int, float]] = [] + for num, vec in self._docs.items(): + if num == issue.number: + continue + score = _cosine(query, vec) + if score > threshold: + scored.append((num, score)) + scored.sort(key=lambda x: x[1], reverse=True) + return scored[:top_k] diff --git a/tools/intel_cpu_triage/github_client.py b/tools/intel_cpu_triage/github_client.py new file mode 100644 index 0000000000000..76e7263c45ed4 --- /dev/null +++ b/tools/intel_cpu_triage/github_client.py @@ -0,0 +1,99 @@ +"""Minimal GitHub API client built on the standard library only. + +Supports the issue search endpoint used for Phase 1 ingestion. Network access +is encapsulated here so the rest of the toolkit can be unit-tested with a fake +client (see ``tests/``). +""" + +from __future__ import annotations + +import json +import time +import urllib.error +import urllib.parse +import urllib.request +from typing import Any, Iterator, Protocol + + +API_ROOT = "https://api.github.com" +USER_AGENT = "intel-cpu-triage/0.1" + + +class SearchClient(Protocol): + """Protocol implemented by both the real and fake GitHub clients.""" + + def search_issues(self, query: str, per_page: int = 100) -> Iterator[dict[str, Any]]: + ... + + +class GitHubClient: + """Tiny REST client. Uses a token from the caller for authentication.""" + + def __init__( + self, + token: str | None = None, + api_root: str = API_ROOT, + max_retries: int = 3, + sleep: float = 2.0, + ) -> None: + self.token = token + self.api_root = api_root.rstrip("/") + self.max_retries = max_retries + self.sleep = sleep + + def _headers(self) -> dict[str, str]: + headers = { + "Accept": "application/vnd.github+json", + "User-Agent": USER_AGENT, + "X-GitHub-Api-Version": "2022-11-28", + } + if self.token: + headers["Authorization"] = "Bearer " + self.token + return headers + + def _get(self, url: str) -> dict[str, Any]: + last_err: Exception | None = None + for attempt in range(self.max_retries): + req = urllib.request.Request(url, headers=self._headers()) + try: + with urllib.request.urlopen(req) as resp: + return json.loads(resp.read().decode("utf-8")) + except urllib.error.HTTPError as exc: + # Respect secondary rate limits / transient 5xx with backoff. + if exc.code in (403, 429, 500, 502, 503) and attempt + 1 < self.max_retries: + time.sleep(self.sleep * (attempt + 1)) + last_err = exc + continue + raise + except urllib.error.URLError as exc: + last_err = exc + time.sleep(self.sleep * (attempt + 1)) + assert last_err is not None + raise last_err + + def search_issues( + self, query: str, per_page: int = 100 + ) -> Iterator[dict[str, Any]]: + """Yield issue items matching a GitHub search query, paginating fully. + + The search API caps results at 1000 items per query; callers should + keep queries narrow (label + updated watermark) to stay under it. + """ + page = 1 + while True: + params = urllib.parse.urlencode( + {"q": query, "per_page": per_page, "page": page} + ) + url = f"{self.api_root}/search/issues?{params}" + data = self._get(url) + items = data.get("items", []) + if not items: + break + for item in items: + # Exclude PRs; the search/issues endpoint returns both. + if "pull_request" in item: + continue + yield item + if len(items) < per_page: + break + page += 1 diff --git a/tools/intel_cpu_triage/ingest.py b/tools/intel_cpu_triage/ingest.py new file mode 100644 index 0000000000000..2f55bb8ad3a82 --- /dev/null +++ b/tools/intel_cpu_triage/ingest.py @@ -0,0 +1,101 @@ +"""Phase 1: incremental ingestion of relevant issues into the local DB. + +Builds GitHub search queries from the configured scope labels and content +keywords, fetches matching open issues updated since the last run (the +"watermark"), and upserts them. The watermark makes each run cheap. +""" + +from __future__ import annotations + +import time +from datetime import datetime, timezone +from typing import Any, Iterator + +from .config import Config +from .db import Database, Issue +from .github_client import SearchClient + + +WATERMARK_KEY = "ingest_watermark" + + +def build_queries(cfg: Config, since: str | None) -> list[str]: + """Build the set of search queries for this run. + + One query per scope label keeps each result set small (under the search + API's 1000-item cap). A final query covers content keywords for issues that + are in-scope by content but mislabeled. + """ + base = f"repo:{cfg.repo} is:issue is:open" + if since: + base += f" updated:>={since}" + + queries: list[str] = [] + for label in cfg.scope_labels: + queries.append(f'{base} label:"{label}"') + + # Content-keyword sweep. GitHub search OR-combines quoted terms; keep the + # list modest to avoid an overly long query string. + if cfg.content_keywords: + terms = " OR ".join(f'"{kw}"' for kw in cfg.content_keywords[:20]) + queries.append(f"{base} in:title,body ({terms})") + return queries + + +def _parse_issue(item: dict[str, Any]) -> Issue: + labels = [ + lbl["name"] if isinstance(lbl, dict) else str(lbl) + for lbl in item.get("labels", []) + ] + reactions = 0 + if isinstance(item.get("reactions"), dict): + reactions = item["reactions"].get("total_count", 0) + user = item.get("user") or {} + return Issue( + number=item["number"], + title=item.get("title", ""), + body=item.get("body") or "", + state=item.get("state", "open"), + author=user.get("login", ""), + author_association=item.get("author_association", ""), + labels=labels, + reactions=reactions, + comments=item.get("comments", 0), + linked_prs=[], + created_at=item.get("created_at", ""), + updated_at=item.get("updated_at", ""), + fetched_at=time.time(), + ) + + +def ingest( + db: Database, + client: SearchClient, + cfg: Config, + now: datetime | None = None, +) -> list[int]: + """Run one incremental ingestion pass. + + Returns the list of issue numbers that were inserted or updated. The + watermark is advanced to the start time of this run so the next pass only + sees newer activity. + """ + now = now or datetime.now(timezone.utc) + run_started = now.strftime("%Y-%m-%dT%H:%M:%SZ") + since = db.get_meta(WATERMARK_KEY) + + seen: dict[int, Issue] = {} + for query in build_queries(cfg, since): + for item in client.search_issues(query): + # Defensive: the search/issues endpoint also returns PRs. + if "pull_request" in item: + continue + issue = _parse_issue(item) + # De-dup across queries within a single run; keep the richest copy. + seen[issue.number] = issue + + for issue in seen.values(): + db.upsert_issue(issue) + + db.set_meta(WATERMARK_KEY, run_started) + return sorted(seen.keys()) diff --git a/tools/intel_cpu_triage/intel_cpu_triage.yml.example b/tools/intel_cpu_triage/intel_cpu_triage.yml.example new file mode 100644 index 0000000000000..291dd74acb662 --- /dev/null +++ b/tools/intel_cpu_triage/intel_cpu_triage.yml.example @@ -0,0 +1,53 @@ +# Example scheduled workflow for Phase 1 ingestion + Phase 2 triage. +# +# Copy this file into your PRIVATE tracking repo as +# .github/workflows/intel_cpu_triage.yml. It is intentionally shipped with a +# .example suffix so it never auto-runs from a pytorch fork. +# +# It runs the heuristic triager out of the box. To use an LLM, replace the +# `triage` step with a small driver script that constructs triage.LLMTriager +# with your model client. + +name: intel-cpu-triage + +on: + schedule: + - cron: "0 7 * * *" # daily at 07:00 UTC + workflow_dispatch: {} + +permissions: + contents: read + +jobs: + triage: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + + # Persist the SQLite DB between runs so ingestion stays incremental. + - name: Restore triage DB + uses: actions/cache@v4 + with: + path: intel_cpu_triage.db + key: intel-cpu-triage-db + + - name: Ingest relevant issues (Phase 1) + env: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + run: | + python -m tools.intel_cpu_triage.cli \ + --config tools/intel_cpu_triage/config.example.json ingest + + - name: Triage & score (Phase 2) + run: | + python -m tools.intel_cpu_triage.cli \ + --config tools/intel_cpu_triage/config.example.json triage + + - name: Report ready-to-work + metrics + run: | + python -m tools.intel_cpu_triage.cli ready --limit 25 + python -m tools.intel_cpu_triage.cli metrics diff --git a/tools/intel_cpu_triage/metrics.py b/tools/intel_cpu_triage/metrics.py new file mode 100644 index 0000000000000..2625b3a01fbb7 --- /dev/null +++ b/tools/intel_cpu_triage/metrics.py @@ -0,0 +1,101 @@ +"""Phase 7: health metrics and the feedback loop. + +Computes metrics that reflect workflow *health* rather than vanity volume: +merge/acceptance rate, change-request (churn) rate, and triage precision. Also +exports labelled examples of accepted vs. rejected PRs so they can be fed back +into the Phase 2 scoring prompt as few-shot guidance. +""" + +from __future__ import annotations + +from dataclasses import dataclass, asdict +from typing import Any + +from .db import BUCKET_READY, Database + + +@dataclass +class Metrics: + total_issues: int + triaged_issues: int + ready_to_work: int + open_prs: int + merged_prs: int + closed_unmerged_prs: int + merge_rate: float # merged / (merged + closed_unmerged) + change_request_rate: float # PRs with changes requested / total reviewed + triage_precision: float # ready issues that progressed past investigating + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +def compute_metrics(db: Database) -> Metrics: + issues = db.all_issues() + prs = db.all_prs() + + triaged = [i for i in issues if i.bucket is not None] + ready = [i for i in issues if i.bucket == BUCKET_READY] + + merged = [p for p in prs if p["state"] == "merged"] + closed_unmerged = [p for p in prs if p["state"] == "closed"] + open_prs = [p for p in prs if p["state"] == "open"] + + decided = len(merged) + len(closed_unmerged) + merge_rate = len(merged) / decided if decided else 0.0 + + reviewed = [p for p in prs if p["state"] in ("merged", "closed")] + churned = [p for p in reviewed if p.get("changes_requested")] + change_request_rate = len(churned) / len(reviewed) if reviewed else 0.0 + + # Triage precision proxy: of issues triaged ready-to-work, how many actually + # advanced beyond the backlog (i.e. a human agreed they were workable)? + progressed = [ + i + for i in ready + if i.board_column not in ("backlog", "closed") + ] + triage_precision = len(progressed) / len(ready) if ready else 0.0 + + return Metrics( + total_issues=len(issues), + triaged_issues=len(triaged), + ready_to_work=len(ready), + open_prs=len(open_prs), + merged_prs=len(merged), + closed_unmerged_prs=len(closed_unmerged), + merge_rate=round(merge_rate, 3), + change_request_rate=round(change_request_rate, 3), + triage_precision=round(triage_precision, 3), + ) + + +def feedback_examples(db: Database, limit: int = 10) -> dict[str, list[dict[str, Any]]]: + """Export accepted/rejected PR outcomes for few-shot scoring feedback. + + Returns ``{"good": [...], "bad": [...]}`` where each entry pairs an issue's + triage record with the eventual PR outcome. Feeding these back into the + Phase 2 prompt improves triage precision over time. + """ + prs = db.all_prs() + good: list[dict[str, Any]] = [] + bad: list[dict[str, Any]] = [] + for pr in prs: + issue_number = pr.get("issue_number") + if issue_number is None: + continue + issue = db.get_issue(issue_number) + if issue is None or issue.triage is None: + continue + example = { + "issue": issue.number, + "title": issue.title, + "triage": issue.triage, + "outcome": pr["state"], + "changes_requested": bool(pr.get("changes_requested")), + } + if pr["state"] == "merged" and not pr.get("changes_requested"): + good.append(example) + elif pr["state"] == "closed": + bad.append(example) + return {"good": good[:limit], "bad": bad[:limit]} diff --git a/tools/intel_cpu_triage/ratelimit.py b/tools/intel_cpu_triage/ratelimit.py new file mode 100644 index 0000000000000..5f13e870f5df3 --- /dev/null +++ b/tools/intel_cpu_triage/ratelimit.py @@ -0,0 +1,54 @@ +"""Phase 6: rate-limiting to protect reviewer trust. + +Enforces a team-wide cap on outstanding external PRs and a per-reviewer budget +so the team never floods maintainers. ``can_open_pr`` is the gate that the +Phase 5 internal-review step should consult before a PR is opened upstream. +""" + +from __future__ import annotations + +from collections import Counter +from dataclasses import dataclass + +from .config import Config +from .db import Database + + +@dataclass +class RateDecision: + allowed: bool + reasons: list[str] + + def __bool__(self) -> bool: + return self.allowed + + +def open_pr_count(db: Database) -> int: + return len(db.open_prs()) + + +def reviewer_load(db: Database) -> dict[str, int]: + counts: Counter[str] = Counter() + for pr in db.open_prs(): + reviewer = pr.get("reviewer") + if reviewer: + counts[reviewer] += 1 + return dict(counts) + + +def can_open_pr(db: Database, cfg: Config, reviewer: str | None = None) -> RateDecision: + """Decide whether opening one more upstream PR is within policy.""" + reasons: list[str] = [] + n_open = open_pr_count(db) + if n_open >= cfg.max_open_external_prs: + reasons.append( + f"team-wide open-PR cap reached ({n_open}/{cfg.max_open_external_prs})" + ) + if reviewer: + load = reviewer_load(db).get(reviewer, 0) + if load >= cfg.max_prs_per_reviewer: + reasons.append( + f"reviewer {reviewer!r} budget reached " + f"({load}/{cfg.max_prs_per_reviewer})" + ) + return RateDecision(allowed=not reasons, reasons=reasons) diff --git a/tools/intel_cpu_triage/tests/__init__.py b/tools/intel_cpu_triage/tests/__init__.py new file mode 100644 index 0000000000000..5944c1bdd405a --- /dev/null +++ b/tools/intel_cpu_triage/tests/__init__.py @@ -0,0 +1,5 @@ +"""Tests for the Intel CPU triage toolkit. + +Run with: python -m pytest tools/intel_cpu_triage/tests + or: python -m unittest discover -s tools/intel_cpu_triage/tests +""" diff --git a/tools/intel_cpu_triage/tests/test_toolkit.py b/tools/intel_cpu_triage/tests/test_toolkit.py new file mode 100644 index 0000000000000..974f155284a8b --- /dev/null +++ b/tools/intel_cpu_triage/tests/test_toolkit.py @@ -0,0 +1,332 @@ +"""Unit tests for the Intel CPU triage toolkit. + +These use stdlib ``unittest`` (no PyTorch build required) since the toolkit is +dependency-free and runs independently of the framework. +""" + +from __future__ import annotations + +import unittest +from datetime import datetime, timezone + +from tools.intel_cpu_triage.config import Config +from tools.intel_cpu_triage.db import ( + BUCKET_NEEDS_DECISION, + BUCKET_NEEDS_INFO, + BUCKET_READY, + Database, + Issue, +) +from tools.intel_cpu_triage import board, ingest, metrics, ratelimit +from tools.intel_cpu_triage.dedup import SimilarityIndex +from tools.intel_cpu_triage.triage import ( + HeuristicTriager, + LLMTriager, + _extract_json, +) + + +def make_issue(number=1, title="t", body="", labels=None, **kw) -> Issue: + return Issue( + number=number, + title=title, + body=body, + labels=labels or [], + **kw, + ) + + +class FakeSearchClient: + """In-memory stand-in for GitHubClient used by ingestion tests.""" + + def __init__(self, items_by_query=None, items=None): + self._items_by_query = items_by_query or {} + self._items = items or [] + + def search_issues(self, query, per_page=100): + if query in self._items_by_query: + yield from self._items_by_query[query] + else: + yield from self._items + + +class ConfigTest(unittest.TestCase): + def test_defaults_present(self): + cfg = Config() + self.assertIn("module: cpu", cfg.scope_labels) + self.assertIn("mkldnn", cfg.content_keywords) + self.assertEqual(cfg.max_open_external_prs, 5) + + +class DatabaseTest(unittest.TestCase): + def setUp(self): + self.db = Database(":memory:") + + def test_upsert_and_get(self): + self.db.upsert_issue(make_issue(1, "hello", labels=["module: cpu"])) + got = self.db.get_issue(1) + self.assertIsNotNone(got) + self.assertEqual(got.title, "hello") + self.assertEqual(got.labels, ["module: cpu"]) + + def test_upsert_preserves_triage_on_reingest(self): + self.db.upsert_issue(make_issue(1, "hello")) + self.db.save_triage(1, BUCKET_READY, 80, 0.9, {"x": 1}) + # Re-ingest (Phase 1) with a fresh Issue that has no triage fields. + self.db.upsert_issue(make_issue(1, "hello updated")) + got = self.db.get_issue(1) + self.assertEqual(got.title, "hello updated") + self.assertEqual(got.bucket, BUCKET_READY) + self.assertEqual(got.actionability, 80) + + def test_board_preserved_on_reingest(self): + self.db.upsert_issue(make_issue(1)) + self.db.save_triage(1, BUCKET_READY, 80, 0.9, {}) + self.db.set_board(1, "investigating", assignee="alice") + self.db.upsert_issue(make_issue(1, "new title")) + got = self.db.get_issue(1) + self.assertEqual(got.board_column, "investigating") + self.assertEqual(got.assignee, "alice") + + def test_meta_watermark(self): + self.assertIsNone(self.db.get_meta("k")) + self.db.set_meta("k", "v") + self.assertEqual(self.db.get_meta("k"), "v") + + +class IngestTest(unittest.TestCase): + def setUp(self): + self.db = Database(":memory:") + self.cfg = Config() + + def test_build_queries_includes_labels_and_keywords(self): + qs = ingest.build_queries(self.cfg, since=None) + self.assertTrue(any('label:"module: cpu"' in q for q in qs)) + self.assertTrue(any("in:title,body" in q for q in qs)) + + def test_build_queries_watermark(self): + qs = ingest.build_queries(self.cfg, since="2024-01-01T00:00:00Z") + self.assertTrue(all("updated:>=2024-01-01T00:00:00Z" in q for q in qs)) + + def test_ingest_dedup_and_watermark(self): + item = { + "number": 42, + "title": "AMX brgemm segfault", + "body": "import torch\nrepro here", + "state": "open", + "labels": [{"name": "module: cpu"}], + "user": {"login": "bob"}, + "author_association": "NONE", + "comments": 3, + "reactions": {"total_count": 5}, + "created_at": "2024-01-01T00:00:00Z", + "updated_at": "2024-02-01T00:00:00Z", + } + client = FakeSearchClient(items=[item]) + now = datetime(2024, 3, 1, tzinfo=timezone.utc) + changed = ingest.ingest(self.db, client, self.cfg, now=now) + self.assertEqual(changed, [42]) + got = self.db.get_issue(42) + self.assertEqual(got.reactions, 5) + self.assertEqual(got.author, "bob") + self.assertEqual(self.db.get_meta(ingest.WATERMARK_KEY), "2024-03-01T00:00:00Z") + + def test_ingest_skips_pull_requests(self): + item = {"number": 1, "title": "pr", "pull_request": {}, "labels": []} + client = FakeSearchClient(items=[item]) + changed = ingest.ingest(self.db, client, self.cfg) + self.assertEqual(changed, []) + + +class HeuristicTriageTest(unittest.TestCase): + def setUp(self): + self.cfg = Config() + self.triager = HeuristicTriager(self.cfg) + + def test_ready_when_scope_and_repro(self): + issue = make_issue( + 1, + "mkldnn conv wrong result", + body="import torch\nSteps to reproduce:\n```python\n...\n```", + labels=["module: mkldnn", "triaged", "high priority"], + reactions=4, + ) + res = self.triager.triage(issue) + self.assertEqual(res.bucket, BUCKET_READY) + self.assertGreaterEqual(res.actionability, self.cfg.ready_score_threshold) + self.assertIn("aten/src/ATen/native/mkldnn/", res.suspected_subsystems) + + def test_needs_info_when_no_repro(self): + issue = make_issue( + 2, "cpu slow", body="it is slow", labels=["module: cpu"] + ) + res = self.triager.triage(issue) + self.assertEqual(res.bucket, BUCKET_NEEDS_INFO) + + def test_quarantine_feature_request(self): + issue = make_issue( + 3, + "Add new API for bf16", + body="import torch\nrepro\nplease add a flag", + labels=["module: cpu", "feature"], + ) + res = self.triager.triage(issue) + self.assertEqual(res.bucket, BUCKET_NEEDS_DECISION) + self.assertTrue(res.quarantine_reasons) + + def test_quarantine_existing_pr(self): + issue = make_issue( + 4, + "mkldnn bug", + body="import torch\n```python\nrepro\n```", + labels=["module: cpu"], + linked_prs=[999], + ) + res = self.triager.triage(issue) + self.assertEqual(res.bucket, BUCKET_NEEDS_DECISION) + self.assertIn("already has a linked/open PR", res.quarantine_reasons) + + def test_content_keyword_in_scope_when_mislabeled(self): + issue = make_issue( + 5, + "Segfault on Sapphire Rapids", + body="import torch\n```python\nAVX512 brgemm crash\n```", + labels=[], # mislabeled + ) + res = self.triager.triage(issue) + self.assertTrue(res.in_scope) + + +class LLMTriageTest(unittest.TestCase): + def setUp(self): + self.cfg = Config() + + def test_llm_result_merged(self): + class FakeLLM: + def complete(self, prompt): + return ( + '{"in_scope": true, "has_repro": true, "fix_type": "kernel",' + ' "difficulty": "small", "suspected_subsystems": ["x/"],' + ' "root_cause_hypothesis": "race", "actionability": 88,' + ' "confidence": 0.8, "quarantine_reasons": []}' + ) + + triager = LLMTriager(self.cfg, FakeLLM()) + issue = make_issue(1, "mkldnn", body="import torch", labels=["module: cpu"]) + res = triager.triage(issue) + self.assertEqual(res.source, "llm") + self.assertEqual(res.actionability, 88) + self.assertEqual(res.bucket, BUCKET_READY) + self.assertEqual(res.root_cause_hypothesis, "race") + + def test_llm_failure_falls_back(self): + class BrokenLLM: + def complete(self, prompt): + raise RuntimeError("no model") + + triager = LLMTriager(self.cfg, BrokenLLM()) + issue = make_issue(1, "mkldnn", body="x", labels=["module: cpu"]) + res = triager.triage(issue) + self.assertEqual(res.source, "heuristic") + + def test_extract_json_with_fences(self): + raw = "```json\n{\"a\": 1}\n```" + self.assertEqual(_extract_json(raw), {"a": 1}) + + +class DedupTest(unittest.TestCase): + def test_similar_issues_rank_high(self): + issues = [ + make_issue(1, "mkldnn conv segfault on avx512", body="brgemm crash amx"), + make_issue(2, "mkldnn convolution crash avx512", body="brgemm segfault amx"), + make_issue(3, "docs typo in readme", body="fix spelling"), + ] + index = SimilarityIndex() + index.build(issues) + ranked = index.most_similar(issues[0], top_k=2) + self.assertEqual(ranked[0][0], 2) + self.assertGreater(ranked[0][1], 0.0) + + +class BoardTest(unittest.TestCase): + def setUp(self): + self.db = Database(":memory:") + self.cfg = Config() + + def _ready_issue(self, number=1): + self.db.upsert_issue(make_issue(number)) + self.db.save_triage(number, BUCKET_READY, 80, 0.9, {}) + + def test_pull_requires_ready(self): + self.db.upsert_issue(make_issue(1)) + self.db.save_triage(1, BUCKET_NEEDS_INFO, 30, 0.5, {}) + with self.assertRaises(board.BoardError): + board.pull_to_board(self.db, 1, "alice") + + def test_pull_and_advance_flow(self): + self._ready_issue(1) + board.pull_to_board(self.db, 1, "alice") + self.assertEqual(self.db.get_issue(1).board_column, "investigating") + board.advance(self.db, 1, "repro_confirmed") + board.advance(self.db, 1, "fix_in_progress") + board.advance(self.db, 1, "pr_open") + board.advance(self.db, 1, "merged") + self.assertEqual(self.db.get_issue(1).board_column, "merged") + + def test_illegal_transition_rejected(self): + self._ready_issue(1) + board.pull_to_board(self.db, 1, "alice") + with self.assertRaises(board.BoardError): + board.advance(self.db, 1, "merged") # skipped steps + + +class RateLimitTest(unittest.TestCase): + def setUp(self): + self.db = Database(":memory:") + self.cfg = Config(max_open_external_prs=2, max_prs_per_reviewer=1) + + def test_team_cap(self): + self.db.upsert_pr(1, state="open") + self.db.upsert_pr(2, state="open") + decision = ratelimit.can_open_pr(self.db, self.cfg) + self.assertFalse(decision) + self.assertTrue(any("cap reached" in r for r in decision.reasons)) + + def test_reviewer_budget(self): + self.db.upsert_pr(1, reviewer="maint", state="open") + decision = ratelimit.can_open_pr(self.db, self.cfg, reviewer="maint") + self.assertFalse(decision) + + def test_allowed_when_under_caps(self): + decision = ratelimit.can_open_pr(self.db, self.cfg, reviewer="maint") + self.assertTrue(decision) + + +class MetricsTest(unittest.TestCase): + def setUp(self): + self.db = Database(":memory:") + + def test_metrics_compute(self): + self.db.upsert_issue(make_issue(1)) + self.db.save_triage(1, BUCKET_READY, 80, 0.9, {}) + self.db.set_board(1, "investigating", assignee="a") + self.db.upsert_pr(1, issue_number=1, state="merged", changes_requested=0) + self.db.upsert_pr(2, issue_number=1, state="closed", changes_requested=1) + m = metrics.compute_metrics(self.db) + self.assertEqual(m.total_issues, 1) + self.assertEqual(m.merged_prs, 1) + self.assertEqual(m.merge_rate, 0.5) + self.assertEqual(m.change_request_rate, 0.5) + self.assertEqual(m.triage_precision, 1.0) + + def test_feedback_examples(self): + self.db.upsert_issue(make_issue(1, "good one")) + self.db.save_triage(1, BUCKET_READY, 80, 0.9, {"k": "v"}) + self.db.upsert_pr(1, issue_number=1, state="merged", changes_requested=0) + fb = metrics.feedback_examples(self.db) + self.assertEqual(len(fb["good"]), 1) + self.assertEqual(fb["good"][0]["issue"], 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/intel_cpu_triage/triage.py b/tools/intel_cpu_triage/triage.py new file mode 100644 index 0000000000000..b24dfab01b2ba --- /dev/null +++ b/tools/intel_cpu_triage/triage.py @@ -0,0 +1,299 @@ +"""Phase 2: AI triage & scoring. + +Each issue is converted into a structured :class:`TriageResult` (serialised to +JSON in the DB). The :class:`LLMTriager` delegates to a user-supplied language +model that must return JSON; if no model is configured or it fails, the +:class:`HeuristicTriager` produces a deterministic, explainable result from +labels and content keywords so the pipeline always makes progress. + +Scoring deliberately *down-ranks or quarantines*: + * issues that already have an open PR (avoid reviewer flooding), + * feature requests / design discussions (need maintainer buy-in), + * anything implying API/semantics changes (need an RFC first). +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field, asdict +from typing import Any, Callable, Protocol + +from .config import Config +from .db import BUCKET_NEEDS_DECISION, BUCKET_NEEDS_INFO, BUCKET_READY, Issue + + +# Phrases that suggest an API/semantics change needing maintainer agreement. +_API_CHANGE_HINTS = ( + "new api", + "add a flag", + "add an option", + "change the default", + "change default", + "deprecate", + "rename", + "signature", + "bc-breaking", + "backward incompat", + "rfc", +) + +_REPRO_HINTS = ( + "import torch", + "```python", + "```py", + "traceback", + "to reproduce", + "steps to reproduce", + "minimal example", + "minimal repro", + "reproduce:", +) + + +@dataclass +class TriageResult: + """Structured triage record (JSON-serialisable).""" + + number: int + bucket: str + actionability: int # 0-100 + confidence: float # 0-1 + in_scope: bool + has_repro: bool + fix_type: str # kernel | numerical | perf | build | docs | unknown + difficulty: str # small | medium | large + suspected_subsystems: list[str] = field(default_factory=list) + root_cause_hypothesis: str = "" + quarantine_reasons: list[str] = field(default_factory=list) + related_issues: list[int] = field(default_factory=list) + rationale: str = "" + source: str = "heuristic" # heuristic | llm + + def to_dict(self) -> dict[str, Any]: + return asdict(self) + + +class LLM(Protocol): + """A language model that returns a JSON string for a given prompt.""" + + def complete(self, prompt: str) -> str: + ... + + +def _matches(text: str, needles) -> list[str]: + low = text.lower() + return [n for n in needles if n.lower() in low] + + +def _detect_quarantine(issue: Issue, cfg: Config) -> list[str]: + reasons: list[str] = [] + if issue.linked_prs: + reasons.append("already has a linked/open PR") + labelset = {label.lower() for label in issue.labels} + if {"feature", "enhancement", "proposal accepted"} & labelset: + reasons.append("feature request / design discussion") + if _matches(issue.text, _API_CHANGE_HINTS): + reasons.append("implies API/semantics change (needs RFC/maintainer)") + return reasons + + +def _detect_subsystems(issue: Issue, cfg: Config) -> list[str]: + found: list[str] = [] + low = issue.text.lower() + for keyword, path in cfg.subsystem_hints.items(): + if keyword in low and path not in found: + found.append(path) + return found + + +def _fix_type(issue: Issue) -> str: + low = issue.text.lower() + labelset = {label.lower() for label in issue.labels} + if "module: docs" in labelset or "documentation" in low: + return "docs" + if "module: build" in labelset or "compil" in low or "cmake" in low: + return "build" + if "module: performance" in labelset or "slow" in low or "regression" in low: + return "perf" + if "nan" in low or "incorrect result" in low or "numerical" in low: + return "numerical" + if any(k in low for k in ("segfault", "crash", "kernel", "brgemm", "assert")): + return "kernel" + return "unknown" + + +class HeuristicTriager: + """Deterministic, explainable triage from labels + content. + + Used as a fallback and as a sane default in offline/CI environments. + """ + + def __init__(self, cfg: Config) -> None: + self.cfg = cfg + + def triage(self, issue: Issue, related: list[int] | None = None) -> TriageResult: + cfg = self.cfg + labelset = {label.lower() for label in issue.labels} + scope_hit = bool({s.lower() for s in cfg.scope_labels} & labelset) + content_hits = _matches(issue.text, cfg.content_keywords) + in_scope = scope_hit or bool(content_hits) + + has_repro = bool(_matches(issue.text, _REPRO_HINTS)) + quarantine = _detect_quarantine(issue, cfg) + subsystems = _detect_subsystems(issue, cfg) + + # Score: start from a base, add maintainer signal, repro, scope, + # engagement; subtract for quarantine and missing repro. + score = 30 + if scope_hit: + score += 20 + if content_hits: + score += min(10, 2 * len(content_hits)) + signal_hit = {s.lower() for s in cfg.signal_labels} & labelset + score += 10 * len(signal_hit) + if "high priority" in labelset: + score += 10 + if has_repro: + score += 15 + score += min(10, issue.reactions) + if "needs reproduction" in labelset or not has_repro: + score -= 15 + if quarantine: + score -= 40 + score = max(0, min(100, score)) + + # Bucket assignment. + if quarantine: + bucket = BUCKET_NEEDS_DECISION + elif not has_repro or "needs reproduction" in labelset: + bucket = BUCKET_NEEDS_INFO + elif in_scope and score >= cfg.ready_score_threshold: + bucket = BUCKET_READY + else: + bucket = BUCKET_NEEDS_INFO + + # Confidence is higher when labels and content agree. + agreement = sum([scope_hit, bool(content_hits), has_repro]) + confidence = round(0.4 + 0.2 * agreement, 2) + confidence = min(confidence, 0.95) + + return TriageResult( + number=issue.number, + bucket=bucket, + actionability=score, + confidence=confidence, + in_scope=in_scope, + has_repro=has_repro, + fix_type=_fix_type(issue), + difficulty="small" if "small" in labelset else "medium", + suspected_subsystems=subsystems, + root_cause_hypothesis="", + quarantine_reasons=quarantine, + related_issues=related or [], + rationale=( + f"scope_label={scope_hit}, content_hits={content_hits}, " + f"signals={sorted(signal_hit)}, repro={has_repro}" + ), + source="heuristic", + ) + + +TRIAGE_PROMPT = """\ +You are triaging a GitHub issue for an Intel CPU / oneDNN team working on +pytorch/pytorch. Respond with a single JSON object and nothing else, using +exactly these keys: + in_scope (bool), has_repro (bool), fix_type (one of + kernel|numerical|perf|build|docs|unknown), difficulty (small|medium|large), + suspected_subsystems (list of repo paths), root_cause_hypothesis (string), + actionability (integer 0-100), confidence (number 0-1), + quarantine_reasons (list of strings). + +Quarantine (low actionability) if the issue already has an open PR, is a +feature request / design discussion, or implies an API/semantics change that +needs an RFC. + +ISSUE #{number} +Labels: {labels} +Title: {title} + +Body: +{body} +""" + + +class LLMTriager: + """Triage via a language model, falling back to heuristics on any failure. + + The model must return JSON. We merge its fields onto a heuristic baseline so + that even a partial/garbled response yields a complete, valid record. The + bucket is *derived* from the merged fields here (not taken from the model) + to keep bucketing policy in one place. + """ + + def __init__(self, cfg: Config, llm: LLM) -> None: + self.cfg = cfg + self.llm = llm + self._fallback = HeuristicTriager(cfg) + + def triage(self, issue: Issue, related: list[int] | None = None) -> TriageResult: + baseline = self._fallback.triage(issue, related) + prompt = TRIAGE_PROMPT.format( + number=issue.number, + labels=", ".join(issue.labels), + title=issue.title, + body=(issue.body or "")[:6000], + ) + try: + raw = self.llm.complete(prompt) + data = _extract_json(raw) + except Exception: + return baseline + if not isinstance(data, dict): + return baseline + + merged = baseline + merged.source = "llm" + for key in ( + "in_scope", + "has_repro", + "fix_type", + "difficulty", + "suspected_subsystems", + "root_cause_hypothesis", + "actionability", + "confidence", + "quarantine_reasons", + ): + if key in data and data[key] is not None: + setattr(merged, key, data[key]) + merged.actionability = max(0, min(100, int(merged.actionability))) + merged.confidence = max(0.0, min(1.0, float(merged.confidence))) + merged.bucket = _bucket_from_result(merged, self.cfg) + return merged + + +def _bucket_from_result(result: TriageResult, cfg: Config) -> str: + if result.quarantine_reasons: + return BUCKET_NEEDS_DECISION + if not result.has_repro: + return BUCKET_NEEDS_INFO + if result.in_scope and result.actionability >= cfg.ready_score_threshold: + return BUCKET_READY + return BUCKET_NEEDS_INFO + + +def _extract_json(raw: str) -> Any: + """Parse JSON from a model response, tolerating code fences/extra prose.""" + raw = raw.strip() + if raw.startswith("```"): + raw = raw.strip("`") + # Drop an optional language tag on the first line. + if "\n" in raw: + first, rest = raw.split("\n", 1) + if first.strip().lower() in ("json", ""): + raw = rest + start = raw.find("{") + end = raw.rfind("}") + if start != -1 and end != -1 and end > start: + raw = raw[start : end + 1] + return json.loads(raw)