#!/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 (last 60 days) and recently closed (last 30 days)5. Asks Claude to sort open candidates into likely and possible duplicates, and surface recently closed issues that may be useful triage context6. Posts a comment if anything is found: a user-facing duplicate alert for likely duplicates, and/or a collapsed triager-facing section for possible duplicates and recently closed related issues 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" # 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=15): """Search issues, returning most recently created first.""" params = {"q": query, "sort": "created", "order": "desc", "per_page": per_page} return github_api_get("/search/issues", params).get("items", []) 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 build_comment(likely_duplicates, possible_duplicates, related_closed_issues): """Compose the full comment body. Returns empty string if there's nothing to post. The comment has two sections, each optional: - User-facing duplicate alert, rendered when likely_duplicates is non-empty. - Collapsed triage context, rendered when there are possible duplicates or related closed issues to surface for triagers. """ sections = [] if likely_duplicates: match_list = "\n".join(f"- #{m['number']}" for m in likely_duplicates) explanations = "\n\n".join( f"**#{m['number']}:** {m['explanation']}\n\n**Shared root cause:** {m['shared_root_cause']}" for m in likely_duplicates ) 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 possible_duplicates or related_closed_issues: parts = [] if possible_duplicates: lines = [ f"- #{m['number']} — {m['explanation']}\n" f" - Possible shared root cause: {m['shared_root_cause']}" for m in possible_duplicates ] parts.append("**Possibly related open issues:**\n\n" + "\n".join(lines)) if related_closed_issues: lines = [ f"- #{m['number']} (closed as {m['state_reason']}) — {m['explanation']}" for m in related_closed_issues ] parts.append("**Recently closed, possibly related:**\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 call_claude(api_key, system, user_content, max_tokens=1024): """Send a message to Claude and return the text response. Raises on non-2xx status.""" 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-sonnet-4-20250514", "max_tokens": max_tokens, "temperature": 0.0, "system": system, "messages": [{"role": "user", "content": user_content}], }, ) 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')}") content = data.get("content", []) if content and content[0].get("type") == "text": return content[0].get("text") or "" return "" 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)) 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. 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 # parse: - [N dupes] https://github.com/.../issues/NUMBER try: dupe_count = int(line.split("[")[1].split()[0]) number = int(line.split("/issues/")[1].split()[0].rstrip(")")) except (ValueError, IndexError): continue # skip "(unlabeled)": these magnets should match everything is_unlabeled = current_area == "(unlabeled)" if number in magnets: if not is_unlabeled: magnets[number]["areas"].append(current_area) else: magnets[number] = { "number": number, "areas": [] if is_unlabeled else [current_area], "dupe_count": dupe_count, } magnet_list = sorted(magnets.values(), key=lambda m: m["dupe_count"], reverse=True) log(f" Parsed {len(magnet_list)} duplicate magnets") return magnet_list def enrich_magnets(magnets): """Fetch title and body_preview for magnets from the API.""" log(f" Fetching details for {len(magnets)} magnets") for magnet in magnets: data = github_api_get(f"/repos/{REPO_OWNER}/{REPO_NAME}/issues/{magnet['number']}") magnet["title"] = data["title"] magnet["body_preview"] = (data.get("body") or "")[:1000] def areas_match(detected, magnet_area): """Check if detected area matches magnet area. Matches broadly across hierarchy levels.""" return ( detected == magnet_area or magnet_area.startswith(f"{detected}/") or detected.startswith(f"{magnet_area}/") ) def filter_magnets_by_areas(magnets, detected_areas): """Filter magnets based on detected areas.""" if not detected_areas: return magnets detected_set = set(detected_areas) def matches(magnet): # unlabeled magnets (empty areas) match everything if not magnet["areas"]: return True return any( areas_match(detected, magnet_area) for detected in detected_set for magnet_area in magnet["areas"] ) return list(filter(matches, magnets)) def search_for_similar_issues(issue, detected_areas, max_searches_per_state=6): """Search for similar issues — both open and recently closed. Runs two passes: - Open issues: title keywords / error pattern unrestricted, area searches last 60 days. - Closed issues: closed within the last 30 days (across all query types). max_searches_per_state caps queries per state to keep token usage and context size bounded. """ log("Searching for similar issues") sixty_days_ago = (datetime.now() - timedelta(days=60)).strftime("%Y-%m-%d") thirty_days_ago = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") title_keywords = [word for word in issue["title"].split() if word.lower() not in STOPWORDS and len(word) > 2] keywords_query = " ".join(title_keywords) if title_keywords else None # error pattern search: capture 5–90 chars after keyword, colon optional error_pattern = r"(?i:\b(?:error|panicked|panic|failed)\b)\s*([^\n]{5,90})" error_match = re.search(error_pattern, issue["body"]) error_snippet = error_match.group(1).strip() if error_match else None def build_queries(base, area_window=None): queries = [] if keywords_query: queries.append(("title_keywords", f"{base} {keywords_query}")) for area in detected_areas: area_q = f'{base} label:"area:{area}"' if area_window: area_q += f" created:>{area_window}" queries.append(("area_label", area_q)) if error_snippet: queries.append(("error_pattern", f'{base} in:body "{error_snippet}"')) return queries open_queries = build_queries( f"repo:{REPO_OWNER}/{REPO_NAME} is:issue is:open", area_window=sixty_days_ago, ) # closed pass: filter by close date so we catch issues closed recently regardless of # when they were opened. closed:> already restricts the result set, so the per-query # area window is unnecessary. closed_queries = build_queries( f"repo:{REPO_OWNER}/{REPO_NAME} is:issue is:closed closed:>{thirty_days_ago}", ) seen_issues = {} for state_label, queries in ( ("open", open_queries[:max_searches_per_state]), ("closed", closed_queries[:max_searches_per_state]), ): for search_type, query in queries: log(f" Search ({state_label} / {search_type}): {query}") try: results = github_search_issues(query, per_page=15) for item in results: number = item["number"] if number != issue["number"] and number not in seen_issues: body = item.get("body") or "" seen_issues[number] = { "number": number, "title": item["title"], "state": item.get("state", ""), "state_reason": item.get("state_reason"), "created_at": item.get("created_at", ""), "body_preview": body[:1000], "source": search_type, } except requests.RequestException as e: log(f" Search failed: {e}") similar_issues = list(seen_issues.values()) log(f" Found {len(similar_issues)} similar issues") return similar_issues def analyze_duplicates(anthropic_key, issue, magnets, search_results): """Use Claude to identify duplicates (open) and surface related closed issues. Returns (likely_duplicates, possible_duplicates, related_closed_issues). """ top_magnets = magnets[:10] magnet_numbers = {m["number"] for m in top_magnets} open_results = [r for r in search_results if r["state"] == "open" and r["number"] not in magnet_numbers] closed_results = [r for r in search_results if r["state"] == "closed" and r["number"] not in magnet_numbers] if not top_magnets and not open_results and not closed_results: return [], [], []Lines 1–500 of 698. Load more
#!/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 taxonomy 3. Parses known "duplicate magnets" from tracking issue #46355 4. Searches for similar issues — open (last 60 days) and recently closed (last 30 days) 5. Asks Claude to sort open candidates into likely and possible duplicates, and surface recently closed issues that may be useful triage context 6. Posts a comment if anything is found: a user-facing duplicate alert for likely duplicates, and/or a collapsed triager-facing section for possible duplicates and recently closed related issues 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 argparse import json import os import re import sys import time from datetime import datetime, timedelta import requests GITHUB_API = "https://api.github.com" REPO_OWNER = "zed-industries" REPO_NAME = "zed" TRACKING_ISSUE_NUMBER = 46355 STAFF_TEAM_SLUG = "staff" # 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 requests TRANSIENT_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=15): """Search issues, returning most recently created first.""" params = {"q": query, "sort": "created", "order": "desc", "per_page": per_page} return github_api_get("/search/issues", params).get("items", []) 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 build_comment(likely_duplicates, possible_duplicates, related_closed_issues): """Compose the full comment body. Returns empty string if there's nothing to post. The comment has two sections, each optional: - User-facing duplicate alert, rendered when likely_duplicates is non-empty. - Collapsed triage context, rendered when there are possible duplicates or related closed issues to surface for triagers. """ sections = [] if likely_duplicates: match_list = "\n".join(f"- #{m['number']}" for m in likely_duplicates) explanations = "\n\n".join( f"**#{m['number']}:** {m['explanation']}\n\n**Shared root cause:** {m['shared_root_cause']}" for m in likely_duplicates ) 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 possible_duplicates or related_closed_issues: parts = [] if possible_duplicates: lines = [ f"- #{m['number']} — {m['explanation']}\n" f" - Possible shared root cause: {m['shared_root_cause']}" for m in possible_duplicates ] parts.append("**Possibly related open issues:**\n\n" + "\n".join(lines)) if related_closed_issues: lines = [ f"- #{m['number']} (closed as {m['state_reason']}) — {m['explanation']}" for m in related_closed_issues ] parts.append("**Recently closed, possibly related:**\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 call_claude(api_key, system, user_content, max_tokens=1024): """Send a message to Claude and return the text response. Raises on non-2xx status.""" 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-sonnet-4-20250514", "max_tokens": max_tokens, "temperature": 0.0, "system": system, "messages": [{"role": "user", "content": user_content}], }, ) 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')}") content = data.get("content", []) if content and content[0].get("type") == "text": return content[0].get("text") or "" return "" 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)) 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. 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 # parse: - [N dupes] https://github.com/.../issues/NUMBER try: dupe_count = int(line.split("[")[1].split()[0]) number = int(line.split("/issues/")[1].split()[0].rstrip(")")) except (ValueError, IndexError): continue # skip "(unlabeled)": these magnets should match everything is_unlabeled = current_area == "(unlabeled)" if number in magnets: if not is_unlabeled: magnets[number]["areas"].append(current_area) else: magnets[number] = { "number": number, "areas": [] if is_unlabeled else [current_area], "dupe_count": dupe_count, } magnet_list = sorted(magnets.values(), key=lambda m: m["dupe_count"], reverse=True) log(f" Parsed {len(magnet_list)} duplicate magnets") return magnet_list def enrich_magnets(magnets): """Fetch title and body_preview for magnets from the API.""" log(f" Fetching details for {len(magnets)} magnets") for magnet in magnets: data = github_api_get(f"/repos/{REPO_OWNER}/{REPO_NAME}/issues/{magnet['number']}") magnet["title"] = data["title"] magnet["body_preview"] = (data.get("body") or "")[:1000] def areas_match(detected, magnet_area): """Check if detected area matches magnet area. Matches broadly across hierarchy levels.""" return ( detected == magnet_area or magnet_area.startswith(f"{detected}/") or detected.startswith(f"{magnet_area}/") ) def filter_magnets_by_areas(magnets, detected_areas): """Filter magnets based on detected areas.""" if not detected_areas: return magnets detected_set = set(detected_areas) def matches(magnet): # unlabeled magnets (empty areas) match everything if not magnet["areas"]: return True return any( areas_match(detected, magnet_area) for detected in detected_set for magnet_area in magnet["areas"] ) return list(filter(matches, magnets)) def search_for_similar_issues(issue, detected_areas, max_searches_per_state=6): """Search for similar issues — both open and recently closed. Runs two passes: - Open issues: title keywords / error pattern unrestricted, area searches last 60 days. - Closed issues: closed within the last 30 days (across all query types). max_searches_per_state caps queries per state to keep token usage and context size bounded. """ log("Searching for similar issues") sixty_days_ago = (datetime.now() - timedelta(days=60)).strftime("%Y-%m-%d") thirty_days_ago = (datetime.now() - timedelta(days=30)).strftime("%Y-%m-%d") title_keywords = [word for word in issue["title"].split() if word.lower() not in STOPWORDS and len(word) > 2] keywords_query = " ".join(title_keywords) if title_keywords else None # error pattern search: capture 5–90 chars after keyword, colon optional error_pattern = r"(?i:\b(?:error|panicked|panic|failed)\b)\s*([^\n]{5,90})" error_match = re.search(error_pattern, issue["body"]) error_snippet = error_match.group(1).strip() if error_match else None def build_queries(base, area_window=None): queries = [] if keywords_query: queries.append(("title_keywords", f"{base} {keywords_query}")) for area in detected_areas: area_q = f'{base} label:"area:{area}"' if area_window: area_q += f" created:>{area_window}" queries.append(("area_label", area_q)) if error_snippet: queries.append(("error_pattern", f'{base} in:body "{error_snippet}"')) return queries open_queries = build_queries( f"repo:{REPO_OWNER}/{REPO_NAME} is:issue is:open", area_window=sixty_days_ago, ) # closed pass: filter by close date so we catch issues closed recently regardless of # when they were opened. closed:> already restricts the result set, so the per-query # area window is unnecessary. closed_queries = build_queries( f"repo:{REPO_OWNER}/{REPO_NAME} is:issue is:closed closed:>{thirty_days_ago}", ) seen_issues = {} for state_label, queries in ( ("open", open_queries[:max_searches_per_state]), ("closed", closed_queries[:max_searches_per_state]), ): for search_type, query in queries: log(f" Search ({state_label} / {search_type}): {query}") try: results = github_search_issues(query, per_page=15) for item in results: number = item["number"] if number != issue["number"] and number not in seen_issues: body = item.get("body") or "" seen_issues[number] = { "number": number, "title": item["title"], "state": item.get("state", ""), "state_reason": item.get("state_reason"), "created_at": item.get("created_at", ""), "body_preview": body[:1000], "source": search_type, } except requests.RequestException as e: log(f" Search failed: {e}") similar_issues = list(seen_issues.values()) log(f" Found {len(similar_issues)} similar issues") return similar_issues def analyze_duplicates(anthropic_key, issue, magnets, search_results): """Use Claude to identify duplicates (open) and surface related closed issues. Returns (likely_duplicates, possible_duplicates, related_closed_issues). """ top_magnets = magnets[:10] magnet_numbers = {m["number"] for m in top_magnets} open_results = [r for r in search_results if r["state"] == "open" and r["number"] not in magnet_numbers] closed_results = [r for r in search_results if r["state"] == "closed" and r["number"] not in magnet_numbers] if not top_magnets and not open_results and not closed_results: return [], [], []