Agent: task-20260401-120519-cluster-ci-failures-by-error-signature-to-deduplic

orchestrator/ci_failure_signatures.py

@@ -0,0 +1,108 @@
+from __future__ import annotations
+
+import re
+
+
+CI_SIGNATURE_SECTION_RE = re.compile(r"^## CI Failure Signature\s*\n(.+?)\s*$", re.MULTILINE)
+_REMAINING_FAILURE_RE = re.compile(r"^## Remaining Failure\s*\n(.*?)(?=^##\s+|\Z)", re.MULTILINE | re.DOTALL)
+_CONTEXT_RE = re.compile(r"^## Context\s*\n(.*?)(?=^##\s+|\Z)", re.MULTILINE | re.DOTALL)
+_FAILED_CHECK_RE = re.compile(r"^\s*-\s+\*\*(.+?)\*\*:", re.MULTILINE)
+_FILE_LINE_PATTERNS: tuple[re.Pattern[str], ...] = (
+    re.compile(r'File "([^"]+)", line (\d+)', re.IGNORECASE),
+    re.compile(r"\b((?:[A-Za-z0-9_.-]+/)+[A-Za-z0-9_.-]+\.[A-Za-z0-9_]+):(\d+)\b"),
+)
+_ERROR_TYPE_RE = re.compile(r"\b([A-Z][A-Za-z0-9_]*(?:Error|Exception|Failure))\b")
+_STACK_FRAME_RE = re.compile(r"\bin ([A-Za-z_][A-Za-z0-9_]*)\b")
+
+
+def extract_ci_failure_signature(title: str, body: str, summary: str = "") -> str | None:
+    explicit = extract_signature_from_body(body)
+    if explicit:
+        return explicit
+
+    text = "\n".join(part for part in [summary, _extract_failure_text(body), _extract_context_text(body), title] if part).strip()
+    if not text:
+        return None
+
+    error_type = _extract_error_type(text)
+    location = _extract_code_location(text)
+    stack_frame = _extract_stack_frame(text)
+    checks = _extract_failed_checks(text)
+
+    anchors = [part for part in [error_type, location, stack_frame] if part]
+    if len(anchors) < 2:
+        return None
+
+    parts = []
+    if checks:
+        parts.append(f"checks={','.join(checks)}")
+    if error_type:
+        parts.append(f"error={error_type}")
+    if location:
+        parts.append(f"location={location}")
+    if stack_frame:
+        parts.append(f"frame={stack_frame}")
+    return " | ".join(parts)
+
+
+def extract_signature_from_body(body: str) -> str | None:
+    match = CI_SIGNATURE_SECTION_RE.search(body or "")
+    if not match:
+        return None
+    return match.group(1).strip() or None
+
+
+def format_signature_section(signature: str | None) -> str:
+    value = str(signature or "").strip()
+    if not value:
+        return ""
+    return f"\n## CI Failure Signature\n{value}\n"
+
+
+def _extract_failure_text(body: str) -> str:
+    match = _REMAINING_FAILURE_RE.search(body or "")
+    if match:
+        return match.group(1).strip()
+    return ""
+
+
+def _extract_context_text(body: str) -> str:
+    match = _CONTEXT_RE.search(body or "")
+    if match:
+        return match.group(1).strip()
+    return ""
+
+
+def _extract_failed_checks(text: str) -> list[str]:
+    seen: set[str] = set()
+    checks: list[str] = []
+    for raw in _FAILED_CHECK_RE.findall(text or ""):
+        normalized = raw.strip().lower()
+        if not normalized or normalized in seen:
+            continue
+        seen.add(normalized)
+        checks.append(normalized)
+    return checks
+
+
+def _extract_error_type(text: str) -> str | None:
+    match = _ERROR_TYPE_RE.search(text or "")
+    if not match:
+        return None
+    return match.group(1)
+
+
+def _extract_code_location(text: str) -> str | None:
+    for pattern in _FILE_LINE_PATTERNS:
+        match = pattern.search(text or "")
+        if not match:
+            continue
+        return f"{match.group(1)}:{match.group(2)}"
+    return None
+
+
+def _extract_stack_frame(text: str) -> str | None:
+    match = _STACK_FRAME_RE.search(text or "")
+    if not match:
+        return None
+    return match.group(1)

orchestrator/github_dispatcher.py

@@ -10,6 +10,7 @@
 
 import yaml
 
+from orchestrator.ci_failure_signatures import extract_ci_failure_signature, extract_signature_from_body
 from orchestrator.paths import load_config, runtime_paths
 from orchestrator.gh_project import (
     get_ready_items,
@@ -59,6 +60,7 @@
     re.compile(r"^\s*-\s+\*\*.+?\*\*:\s*`.+?`\s*(?:- .+)?$", re.IGNORECASE),
 )
 BLOCKED_ESCALATION_COMMENT_MARKER = "<!-- agent-os-blocked-task-escalation -->"
+_DUPLICATE_PARENT_RE = re.compile(r"^## Duplicate CI Signature Parent\s*\n#?(\d+)\s*$", re.MULTILINE)
 
 
 def slugify(text: str) -> str:
@@ -125,6 +127,93 @@ def parse_issue_dependencies(body: str) -> list[int]:
     return deps
 
 
+def _extract_duplicate_parent_issue(body: str) -> int | None:
+    match = _DUPLICATE_PARENT_RE.search(body or "")
+    if not match:
+        return None
+    return int(match.group(1))
+
+
+def _issue_debug_signature(issue: dict) -> str | None:
+    body = issue.get("body", "") or ""
+    parsed = parse_issue_body(body)
+    if parsed.get("task_type") != "debugging":
+        return None
+    return extract_ci_failure_signature(issue.get("title", "") or "", body)
+
+
+def _attach_duplicate_signature_dependency(
+    repo_full: str,
+    dependent_issue: dict,
+    primary_issue: dict,
+    info: dict | None,
+    project_cfg: dict,
+    signature: str,
+):
+    dependent_number = int(dependent_issue["number"])
+    primary_number = int(primary_issue["number"])
+    body = dependent_issue.get("body", "") or ""
+    dependencies = parse_issue_dependencies(body)
+    if dependencies:
+        return
+
+    signature_section = ""
+    if not extract_signature_from_body(body):
+        signature_section = f"\n## CI Failure Signature\n{signature}\n"
+    suffix = (
+        f"\n\nDepends on #{primary_number}\n"
+        f"{signature_section}\n"
+        f"## Duplicate CI Signature Parent\n#{primary_number}\n"
+    )
+    gh([
+        "api",
+        f"repos/{repo_full}/issues/{dependent_number}",
+        "-X", "PATCH",
+        "-f", f"body={body.rstrip()}{suffix}",
+    ], check=False)
+
+    edit_issue_labels(
+        repo_full,
+        dependent_number,
+        add=["blocked"],
+        remove=["ready", "in-progress", "agent-dispatched"],
+    )
+    add_issue_comment(
+        repo_full,
+        dependent_number,
+        (
+            f"Blocked automatically behind #{primary_number} because it matches the same CI failure signature.\n\n"
+            f"`{signature}`"
+        ),
+    )
+    if info is not None and dependent_issue.get("item_id"):
+        _set_project_status(info, dependent_issue["item_id"], project_cfg.get("blocked_value", "Blocked"))
+
+
+def _cluster_duplicate_debug_issues(
+    cfg: dict,
+    repo_full: str,
+    primary_issue: dict,
+    info: dict | None,
+    project_cfg: dict,
+    ready_items: list[dict],
+):
+    signature = _issue_debug_signature(primary_issue)
+    if not signature:
+        return
+
+    for candidate in ready_items:
+        if candidate.get("repo") != repo_full or candidate.get("number") == primary_issue.get("number"):
+            continue
+        if parse_issue_dependencies(candidate.get("body", "") or ""):
+            continue
+        if _extract_duplicate_parent_issue(candidate.get("body", "") or ""):
+            continue
+        if _issue_debug_signature(candidate) != signature:
+            continue
+        _attach_duplicate_signature_dependency(repo_full, candidate, primary_issue, info, project_cfg, signature)
+
+
 def _repo_agent_fallbacks(cfg: dict, project_key: str) -> dict:
     project_cfg = cfg.get("github_projects", {}).get(project_key, {})
     if isinstance(project_cfg, dict):
@@ -1448,6 +1537,8 @@ def _dispatch_item(cfg, paths, owner, repo_to_project, info, ready_items, issue_
                 )
                 continue
 
+        _cluster_duplicate_debug_issues(cfg, repo_full, item, info, pcfg, ready_items)
+
         # --- Task decomposition: split epics into sub-issues ---
         decomp = _try_decompose(cfg, repo_full, item, info, pcfg)
         if decomp is not None:
@@ -1656,6 +1747,7 @@ def dispatch_one():
                                 f"{PUSH_NOT_READY_CODE}: {reason_codes}"
                             )
                             continue
+                    _cluster_duplicate_debug_issues(cfg, repo_full, issue_with_label_set, None, project_cfg, issues)
                     try:
                         task_id, task_md = build_mailbox_task(cfg, project_key, repo_cfg, issue)
                     except ValueError as exc: