#!/usr/bin/env python3"""Comment on newly opened issues with possible duplicates and triage hints. This script is run by a GitHub Actions workflow when a new issue is opened. It:1. Checks eligibility (bug/crash type or untyped, non-staff author)2. Detects relevant areas using Claude + the area label taxonomy3. Parses known "duplicate magnets" from tracking issue #463554. Searches for similar issues — open (including long-lived, recently active issues) and recently   closed (last 90 days) — and Discussions (feature requests / open-ended topics)5. Asks Claude to sort open candidates into likely and possible duplicates, surface   recently closed issues that may be useful triage context, and flag discussions the   issue may duplicate6. Posts a comment if anything is found: a user-facing alert for likely duplicate issues   or discussions, and/or a collapsed triager-facing section for possible duplicates,   recently closed related issues, and possibly-related discussions Requires:    requests (pip install requests) Usage:    python github-check-new-issue-for-duplicates.py <issue_number> Environment variables:    GITHUB_TOKEN       - GitHub token (org members: read, issues: read & write)    ANTHROPIC_API_KEY  - Anthropic API key for Claude """ import argparseimport jsonimport osimport reimport sysimport timefrom datetime import datetime, timedelta import requests GITHUB_API = "https://api.github.com"REPO_OWNER = "zed-industries"REPO_NAME = "zed"TRACKING_ISSUE_NUMBER = 46355STAFF_TEAM_SLUG = "staff"CLAUDE_MODEL = "claude-sonnet-4-6" # area prefixes to collapse in taxonomy (show summary instead of all sub-labels)PREFIXES_TO_COLLAPSE = ["languages", "parity", "tooling"] # stopwords to filter from title keyword searches (short words handled by len > 2 filter)STOPWORDS = {    "after", "all", "also", "and", "any", "but", "can't", "does", "doesn't",    "don't", "for", "from", "have", "just", "not", "only", "some", "that",    "the", "this", "when", "while", "with", "won't", "work", "working", "zed",} # HTTP statuses we'll retry on for GET requestsTRANSIENT_HTTP_STATUSES = {429, 500, 502, 503, 504}  def log(message):    """Print to stderr so it doesn't interfere with JSON output on stdout."""    print(message, file=sys.stderr)  def github_api_get(path, params=None):    """Fetch JSON from the GitHub API, retrying transient failures. Raises on non-2xx status."""    url = f"{GITHUB_API}/{path.lstrip('/')}"    for attempt in range(3):        try:            response = requests.get(url, headers=GITHUB_HEADERS, params=params)            response.raise_for_status()            return response.json()        except requests.RequestException as e:            transient = isinstance(e, (requests.ConnectionError, requests.Timeout)) or (                isinstance(e, requests.HTTPError) and e.response.status_code in TRANSIENT_HTTP_STATUSES            )            if not transient or attempt == 2:                raise            wait = 2 ** attempt            log(f"  Transient GitHub API error ({e}); retrying in {wait}s")            time.sleep(wait)  def github_search_issues(query, per_page=50, sort=None):    """Search issues, using GitHub's relevance ordering unless a sort is specified."""    params = {"q": query, "per_page": per_page}    if sort:        params.update({"sort": sort, "order": "desc"})    return github_api_get("/search/issues", params).get("items", [])  def github_api_graphql(query, variables=None):    """Run a GraphQL query against the GitHub API, retrying transient failures. """    url = f"{GITHUB_API}/graphql"    for attempt in range(3):        try:            response = requests.post(                url, headers=GITHUB_HEADERS, json={"query": query, "variables": variables or {}}            )            response.raise_for_status()            data = response.json()            if "errors" in data:                raise ValueError(f"GraphQL errors: {json.dumps(data['errors'])[:300]}")            return data["data"]        except requests.RequestException as e:            transient = isinstance(e, (requests.ConnectionError, requests.Timeout)) or (                isinstance(e, requests.HTTPError) and e.response.status_code in TRANSIENT_HTTP_STATUSES            )            if not transient or attempt == 2:                raise            wait = 2 ** attempt            log(f"  Transient GitHub GraphQL error ({e}); retrying in {wait}s")            time.sleep(wait)  def check_team_membership(org, team_slug, username):    """Check if user is an active member of a team."""    try:        data = github_api_get(f"/orgs/{org}/teams/{team_slug}/memberships/{username}")        return data.get("state") == "active"    except requests.HTTPError as e:        if e.response.status_code == 404:            return False        raise  def post_comment(issue_number: int, body):    url = f"{GITHUB_API.rstrip('/')}/repos/{REPO_OWNER}/{REPO_NAME}/issues/{issue_number}/comments"    response = requests.post(url, headers=GITHUB_HEADERS, json={"body": body})    response.raise_for_status()    log(f"  Posted comment on #{issue_number}")  def format_candidate_reference(match):    candidate = match["candidate"]    if candidate["kind"] == "discussion":        return f"[Discussion #{candidate['number']}]({candidate['url']})"    return f"#{candidate['number']}"  def build_comment(likely_matches, possible_matches, related_closed_candidates):    """Compose the full comment body. Returns empty string if there's nothing to post."""    sections = []    likely_issues = [m for m in likely_matches if m["candidate"]["kind"] == "issue"]    likely_discussions = [m for m in likely_matches if m["candidate"]["kind"] == "discussion"]     if likely_issues:        match_list = "\n".join(f"- {format_candidate_reference(m)}" for m in likely_issues)        explanations = "\n\n".join(            f"**{format_candidate_reference(m)}:** {m['explanation']}\n\n"            f"**Shared root cause:** {m['shared_root_cause']}"            for m in likely_issues        )        sections.append(f"""This issue appears to be a duplicate of: {match_list} **If this is indeed a duplicate:**Please close this issue and subscribe to the linked issue for updates (select "Close as not planned" → "Duplicate") **If this is a different issue:**No action needed. A maintainer will review this shortly. <details><summary>Why were these issues selected?</summary> {explanations} </details>""")     if likely_discussions:        match_list = "\n".join(f"- {format_candidate_reference(m)}" for m in likely_discussions)        explanations = "\n\n".join(            f"**{format_candidate_reference(m)}:** {m['explanation']}" for m in likely_discussions        )        sections.append(f"""This looks like it may already be covered by an existing discussion: {match_list} Zed tracks feature requests and open-ended topics in Discussions rather than Issues. **If your report is covered there, please close this issue as a duplicate (select "Close as not planned" → "Duplicate") and continue in the discussion** so the conversation stays in one place. <details><summary>Why were these selected?</summary> {explanations} </details>""")     possible_issues = [m for m in possible_matches if m["candidate"]["kind"] == "issue"]    possible_discussions = [m for m in possible_matches if m["candidate"]["kind"] == "discussion"]    if possible_matches or related_closed_candidates:        parts = []        if possible_issues:            lines = [                f"- {format_candidate_reference(m)} — {m['explanation']}\n"                f"  - Possible shared root cause: {m['shared_root_cause']}"                for m in possible_issues            ]            parts.append("**Possibly related open issues:**\n\n" + "\n".join(lines))        if related_closed_candidates:            lines = [                f"- {format_candidate_reference(m)}"                f"{' (closed as duplicate)' if m['candidate'].get('state_reason') == 'duplicate' else ''}"                f" — {m['explanation']}"                for m in related_closed_candidates            ]            parts.append("**Recently closed, possibly the same bug:**\n\n" + "\n".join(lines))        if possible_discussions:            lines = [                f"- {format_candidate_reference(m)} — {m['explanation']}"                for m in possible_discussions            ]            parts.append("**Possibly related discussions:**\n\n" + "\n".join(lines))        body = "\n\n".join(parts)        sections.append(f"""<details><summary>Additional recent context for triagers</summary> {body} </details>""")     if not sections:        return ""     sections.append("---\n<sub>This is an automated analysis and might be incorrect.</sub>")    return "\n\n".join(sections)  def _claude_request(api_key, payload):    """POST to the Claude Messages API, raise on non-2xx, log token usage, return parsed data."""    response = requests.post(        "https://api.anthropic.com/v1/messages",        headers={            "x-api-key": api_key,            "anthropic-version": "2023-06-01",            "content-type": "application/json",        },        json={"model": CLAUDE_MODEL, "temperature": 0.0, **payload},    )    response.raise_for_status()    data = response.json()     usage = data.get("usage", {})    log(f"  Token usage - Input: {usage.get('input_tokens', 'N/A')}, Output: {usage.get('output_tokens', 'N/A')}")    return data  def call_claude(api_key, system_prompt, user_content, max_tokens=1024):    """Send a message to Claude and return the text response. Raises on non-2xx status."""    data = _claude_request(api_key, {        "max_tokens": max_tokens,        "system": system_prompt,        "messages": [{"role": "user", "content": user_content}],    })     content = data.get("content", [])    if content and content[0].get("type") == "text":        return content[0].get("text") or ""    return ""  def call_claude_tool(api_key, system_prompt, user_content, tool, max_tokens=1024):    """Call Claude, forcing it to invoke `tool`, and return the structured input dict.     Forcing a tool call makes the API emit schema-shaped JSON via its tool-use mechanism    instead of free-form text we'd have to parse out of prose or markdown fences. Raises on    non-2xx status, or if no tool_use block is returned.    """    data = _claude_request(api_key, {        "max_tokens": max_tokens,        "system": system_prompt,        "messages": [{"role": "user", "content": user_content}],        "tools": [tool],        "tool_choice": {"type": "tool", "name": tool["name"]},    })     if data.get("stop_reason") == "max_tokens":        log("  Warning: response hit max_tokens; structured output may be truncated")     for block in data.get("content", []):        if block.get("type") == "tool_use":            return block.get("input") or {}    raise ValueError(f"Claude returned no tool_use block for tool '{tool['name']}'")  def fetch_issue(issue_number: int):    """Fetch issue from GitHub and return as a dict."""    log(f"Fetching issue #{issue_number}")     issue_data = github_api_get(f"/repos/{REPO_OWNER}/{REPO_NAME}/issues/{issue_number}")    issue = {        "number": issue_number,        "title": issue_data["title"],        "body": issue_data.get("body") or "",        "author": (issue_data.get("user") or {}).get("login") or "",        "type": (issue_data.get("type") or {}).get("name"),    }     log(f"  Title: {issue['title']}\n  Type: {issue['type']}\n  Author: {issue['author']}")    return issue  def should_skip(issue):    """Check if issue should be skipped in duplicate detection process."""    if issue["type"] and issue["type"] not in ["Bug", "Crash"]:        log(f"  Skipping: issue type '{issue['type']}' is not blank and not a bug/crash report")        return True     if issue["author"] and check_team_membership(REPO_OWNER, STAFF_TEAM_SLUG, issue["author"]):        log(f"  Skipping: author '{issue['author']}' is a {STAFF_TEAM_SLUG} member")        return True     return False  def fetch_area_labels():    """Fetch area:* labels from the repository. Returns list of {name, description} dicts."""    log("Fetching area labels")     labels = []    page = 1    while page_labels := github_api_get(        f"/repos/{REPO_OWNER}/{REPO_NAME}/labels",        params={"per_page": 100, "page": page},    ):        labels.extend(page_labels)        page += 1     # label["name"][5:] removes the "area:" prefix    area_labels = [        {"name": label["name"][5:], "description": label.get("description") or ""}        for label in labels        if label["name"].startswith("area:")    ]     log(f"  Found {len(area_labels)} area labels")    return area_labels  def format_taxonomy_for_claude(area_labels):    """Format area labels into a string for Claude, collapsing certain prefixes."""    lines = set()     for area in area_labels:        name = area["name"]        collapsible_prefix = next(            (p for p in PREFIXES_TO_COLLAPSE if name.startswith(f"{p}/")), None)         if collapsible_prefix:            lines.add(f"- {collapsible_prefix}/* (multiple specific sub-labels exist)")        else:            desc = area["description"]            lines.add(f"- {name}: {desc}" if desc else f"- {name}")     return "\n".join(sorted(lines))  SEARCH_QUERY_TOOL = {    "name": "report_search_queries",    "description": "Report concise GitHub search queries for finding duplicate reports.",    "input_schema": {        "type": "object",        "properties": {            "queries": {                "type": "array",                "items": {"type": "string"},                "minItems": 1,                "maxItems": 3,            },        },        "required": ["queries"],    },}  def generate_search_queries(anthropic_key, issue):    """Use Claude to derive a few concise searches from the issue's title and body."""    log("Generating search queries with Claude")    system_prompt = """Generate 1-3 concise GitHub search queries for finding earlier reports of thesame underlying bug or request. Each query should contain 2-5 terms that are likely to appear in acanonical issue's title or body. Use the report body as well as its title, include distinctive termssuch as error codes or requested mechanisms, and vary vocabulary when useful. Do not include GitHubqualifiers such as repo:, is:, label:, or in:, and do not wrap terms in quotes."""    user_content = f"""# Issue Title{issue['title']} # Issue Body{issue['body'][:4000]}"""    try:        response = call_claude_tool(            anthropic_key,            system_prompt,            user_content,            SEARCH_QUERY_TOOL,            max_tokens=300,        )    except (requests.RequestException, ValueError) as error:        log(f"  Search query generation failed ({error}); falling back to title and area searches")        return []     queries = []    for query in response.get("queries", []):        if not isinstance(query, str) or re.search(            r"\b(?:repo|is|label|state|created|updated|closed|in):", query, re.IGNORECASE        ):            continue        terms = re.findall(r"[A-Za-z0-9][A-Za-z0-9_./+#'-]*", query)        normalized = " ".join(terms[:5])        if normalized and normalized not in queries:            queries.append(normalized)     log(f"  Generated search queries: {queries}")    return queries  def detect_areas(anthropic_key, issue, area_labels):    """Use Claude to detect which area labels apply to the issue.     Claude may ignore the format instruction or hallucinate names, so the response    is validated against the canonical set of area labels.    """    log("Detecting areas with Claude")     taxonomy = format_taxonomy_for_claude(area_labels)    valid_areas = {label["name"] for label in area_labels}     system_prompt = """You analyze GitHub issues to identify which area labels apply. Decide the area from the user's stated symptom and reproduction steps. Issue bodies routinelycontain pasted log output, crash dumps, stack traces, settings files, and template headers like"Attach Zed log file" or "Relevant Zed settings" — these are evidence about the symptom andshould not push you toward labels like "logging" or "settings" unless the bug itself is abouthow that subsystem works. Respond with ONLY a comma-separated list of matching area names. No prose, no explanation,no markdown, no preamble — just the names. - Output at most 3 areas, ranked by relevance- Use exact area names from the taxonomy- If no areas clearly match, respond with: none- For languages/*, tooling/*, or parity/*, use the specific sub-label (e.g., "languages/rust",  tooling/eslint, parity/vscode) Examples of valid responses (each line is a complete response on its own):  editor, parity/vim  ai, ai/agent panel  none"""     user_content = f"""## Area Taxonomy{taxonomy} # Issue Title{issue['title']} # Issue Body{issue['body'][:4000]}"""     response = call_claude(anthropic_key, system_prompt, user_content, max_tokens=100).strip()    log(f"  Detected areas: {response}")     if response.lower() == "none":        return []     valid, dropped = [], []    for area in response.split(","):        area = area.strip()        (valid if area in valid_areas else dropped).append(area)    if dropped:        log(f"  Dropped {len(dropped)} unknown area(s) from Claude response: {dropped}")    return valid  def parse_duplicate_magnets():    """Parse known duplicate magnets from tracking issue #46355.     Returns a list of magnets sorted by duplicate count (most duplicated first).    Magnets only have number, areas, and dupe_count — use enrich_magnets() to fetch    title and body_preview for the ones you need.    """    log(f"Parsing duplicate magnets from #{TRACKING_ISSUE_NUMBER}")     issue_data = github_api_get(f"/repos/{REPO_OWNER}/{REPO_NAME}/issues/{TRACKING_ISSUE_NUMBER}")    body = issue_data.get("body") or ""     # parse the issue body    # format: ## area_name    #         -   [N dupes] https://github.com/zed-industries/zed/issues/NUMBER    magnets = {}  # number -> {number, areas, dupe_count}    current_area = None     for line in body.split("\n"):        # check for area header        if line.startswith("## "):            current_area = line[3:].strip()            continue         if not current_area or not line.startswith("-") or "/issues/" not in line:            continue

Lines 1–500 of 1402. Load more