Failure Deduplication

test_user_login() ❌ ConnectionRefusedError: [Errno 111]
test_user_logout() ❌ ConnectionRefusedError: [Errno 111]
test_user_profile() ❌ ConnectionRefusedError: [Errno 111]
# ... 47 more identical failures

def get_failure_signature(failure: TestFailure) -> str:
    """Create a signature for grouping identical failures.

    Uses error message and first few lines of stack trace to identify
    failures that are essentially the same issue.
    """
    # Use error message and first 5 lines of stack trace
    stack_lines = failure.stack_trace.split("\n")[:5]
    signature_text = f"{failure.error_message}|{'|'.join(stack_lines)}"
    return hashlib.sha256(signature_text.encode()).hexdigest()

# Group failures by signature to avoid analyzing identical errors multiple times
failure_groups: dict[str, list[TestFailure]] = defaultdict(list)
for tf in test_failures:
    sig = get_failure_signature(tf)
    failure_groups[sig].append(tf)

unique_errors = len(failure_groups)
logger.info(
    f"Grouped {len(test_failures)} failures into {unique_errors} unique error types"
)

failure_groups = {
    "a3f5d8e...": [test1, test2, test3],  # ConnectionRefusedError group
    "9b2c1f7...": [test4, test5],         # AssertionError group
    "c4e6a9d...": [test6],                # Unique timeout
}

# Analyze each unique failure group in parallel
failure_tasks = [
    analyze_failure_group(
        failures=group,
        console_context=console_context,
        repo_path=repo_path,
        ai_provider=ai_provider,
        ai_model=ai_model,
        ai_cli_timeout=settings.ai_cli_timeout,
    )
    for group in failure_groups.values()
]
group_results = await run_parallel_with_limit(failure_tasks)

# Flatten results and handle exceptions
failures = []
group_list = list(failure_groups.values())
for i, result in enumerate(group_results):
    if isinstance(result, Exception):
        # Create error entries for all failures in this group
        for tf in group_list[i]:
            failures.append(
                FailureAnalysis(
                    test_name=tf.test_name,
                    error=tf.error_message,
                    analysis=AnalysisDetail(
                        details=f"Analysis failed: {result}"
                    ),
                )
            )
    else:
        failures.extend(result)

async def analyze_failure_group(
    failures: list[TestFailure],
    console_context: str,
    repo_path: Path | None,
    ai_provider: str = "",
    ai_model: str = "",
    ai_cli_timeout: int | None = None,
) -> list[FailureAnalysis]:
    """Analyze a group of failures with the same error signature.

    Only calls Claude CLI once for the group, then applies the analysis
    to all failures in the group.
    """
    # Use the first failure as representative
    representative = failures[0]
    test_names = [f.test_name for f in failures]

    prompt = f"""Analyze this test failure from a Jenkins CI job.

AFFECTED TESTS ({len(failures)} tests with same error):
{chr(10).join(f"- {name}" for name in test_names)}

ERROR: {representative.error_message}
STACK TRACE:
{representative.stack_trace}

CONSOLE CONTEXT:
{console_context}

You have access to the test repository. Explore the code to understand the failure.

Note: Multiple tests failed with the same error. Provide ONE analysis that applies to all of them.

{_JSON_RESPONSE_SCHEMA}
"""

# Apply the same analysis to all failures in the group
return [
    FailureAnalysis(
        test_name=f.test_name,
        error=f.error_message,
        analysis=parsed,
    )
    for f in failures
]

[
  {"test_name": "test_vm_start", "error": "Timeout waiting for VM", "stack": "...line 45..."},
  {"test_name": "test_vm_stop", "error": "Timeout waiting for VM", "stack": "...line 45..."},
  {"test_name": "test_vm_restart", "error": "Timeout waiting for VM", "stack": "...line 45..."},
  // ... 5 more timeout failures
  {"test_name": "test_network_create", "error": "NullPointerException", "stack": "...line 89..."},
  {"test_name": "test_network_delete", "error": "NullPointerException", "stack": "...line 89..."},
  // ... 2 more NPE failures
  {"test_name": "test_storage_mount", "error": "Permission denied", "stack": "...line 123..."}
]

failure_groups = {
    "hash_a": [test_vm_start, test_vm_stop, test_vm_restart, ...],  # 8 tests
    "hash_b": [test_network_create, test_network_delete, ...],       # 3 tests
    "hash_c": [test_storage_mount],                                  # 1 test
}

total_failures = len(failures)
# Include deduplication info in summary if applicable
if unique_errors > 0 and unique_errors < total_failures:
    summary = (
        f"{total_failures} failure(s) analyzed "
        f"({unique_errors} unique error type(s))"
    )
else:
    summary = f"{total_failures} failure(s) analyzed"

{
  "summary": "12 failure(s) analyzed (3 unique error types)",
  "failures": [
    // ... all 12 failures with analysis attached
  ]
}

# Group failures by signature to avoid analyzing identical errors multiple times
failure_groups: dict[str, list[TestFailure]] = defaultdict(list)
for tf in test_failures:
    sig = get_failure_signature(tf)
    failure_groups[sig].append(tf)

logger.info(
    f"Grouped {len(test_failures)} failures into {len(failure_groups)} unique error types"
)

# Analyze each unique failure group in parallel
tasks = [
    analyze_failure_group(
        failures=group,
        console_context=console_context,
        repo_path=repo_path,
        ai_provider=ai_provider,
        ai_model=ai_model,
        ai_cli_timeout=ai_cli_timeout,
    )
    for group in failure_groups.values()
]
group_results = await run_parallel_with_limit(tasks)

# Deduplicate by keyword set — same keywords = one Jira search
keyword_to_reports: dict[tuple[str, ...], list[ProductBugReport]] = {}
for report in reports:
    if not report.jira_search_keywords:
        continue
    key = tuple(sorted(report.jira_search_keywords))
    keyword_to_reports.setdefault(key, []).append(report)

test_a: "Timeout waiting for VM (network issue)"
test_b: "Timeout waiting for VM (disk full)"

sig_a = hash("Timeout... | network_driver.py:45 | ...")
sig_b = hash("Timeout... | storage_handler.py:89 | ...")

test_retry_1: "AssertionError: expected 5, got 3"
test_retry_2: "AssertionError: expected 5, got 4"

failure.stack_trace = ""

stack_lines = "".split("\n")[:5]  # []
signature_text = f"{failure.error_message}|"  # No stack info

INFO: Grouped 45 failures into 8 unique error types
INFO: Calling CLAUDE CLI for failure group (12 tests with same error)
INFO: Calling CLAUDE CLI for failure group (8 tests with same error)
INFO: Calling CLAUDE CLI for failure group (5 tests with same error)
...

{
  "summary": "45 failure(s) analyzed (8 unique error types)",
  "failures": [...]
}

<h3>CODE ISSUE: API endpoint misconfiguration (12 tests affected)</h3>
<ul>
  <li>test_user_login</li>
  <li>test_user_logout</li>
  <!-- ... 10 more -->
</ul>

MAX_CONCURRENT_AI_CALLS = 10