orchestrata_ultimate.verification.edge_cases.detector

Reference › orchestrata_ultimate › verification › edge_cases › detector

The edge-case detection system (Spec 035 US-006) and the largest module of the edge_cases package. It has four layers.

• Vocabulary: EdgeCaseSeverity (low to critical), EdgeCaseCategory (input validation, boundary condition, error handling, concurrency, performance, security, resource management, state management), and the dataclasses EdgeCase (a catalogue entry with detection patterns, mitigation and frequency), EdgeCaseOccurrence and EdgeCaseCoverage.
• EdgeCaseDatabase: an in-memory catalogue seeded with common edge cases by _init_common_edge_cases, indexed by task type, with occurrence and coverage records. It accepts a storage_path argument.
• EdgeCaseDetector: detect converts verification results into the simpler EdgeCase of types, with partial results as medium severity and errors as high. detect_boundary_cases flags a result whose numeric actual value lies within a tolerance (default 5 percent) of the expected value but is not equal to it. detect_edge_cases applies each applicable catalogue entry's regular expressions line by line to source code, ignoring invalid patterns, and records an occurrence with the matching line range. suggest_edge_cases, calculate_coverage and generate_test_suggestions compare the catalogue with existing test code and propose tests.
• EdgeCaseReporter: generate_report builds a per-task report with recommendations, and export_json and export_markdown render it.

At a glance

Class diagram

Dependency diagram

API

Classes

EdgeCaseSeverity (enum)

Bases: Enum · Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-005']) · Line 24

Severity level for edge cases.

EdgeCaseCategory (enum)

Bases: Enum · Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-005']) · Line 34

Category of edge case.

EdgeCase (dataclass)

Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-008']), dataclass · Line 49

Definition of an edge case.

EdgeCaseOccurrence (dataclass)

Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-002', 'REQ-035-US-006-004']), dataclass · Line 72

An occurrence of an edge case in a specific task.

EdgeCaseCoverage (dataclass)

Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-006']), dataclass · Line 85

Coverage information for edge cases in a task.

EdgeCaseDatabase

Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-009']) · Line 97

Database for storing and retrieving edge cases.

EdgeCaseDetector

Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-001', 'REQ-035-US-006-010']) · Line 380

Detector for identifying edge cases in code.

EdgeCaseReporter

Decorators: traces(spec='035', story='US-006', reqs=['REQ-035-US-006-014']) · Line 681

Reporter for edge case analysis results.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/verification/edge_cases/detector.py

"""Edge Case Detection System for Verification.

This module implements edge case identification, tracking, and suggestion
to help agents produce robust implementations.
"""

import json
import re
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.types import EdgeCase as TypesEdgeCase
from orchestrata_ultimate.verification.types import (
    VerificationResult,
    VerificationStatus,
)


@traces(spec="035", story="US-006", reqs=["REQ-035-US-006-005"])
class EdgeCaseSeverity(Enum):
    """Severity level for edge cases."""

    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@traces(spec="035", story="US-006", reqs=["REQ-035-US-006-005"])
class EdgeCaseCategory(Enum):
    """Category of edge case."""

    INPUT_VALIDATION = "input_validation"
    BOUNDARY_CONDITION = "boundary_condition"
    ERROR_HANDLING = "error_handling"
    CONCURRENCY = "concurrency"
    PERFORMANCE = "performance"
    SECURITY = "security"
    RESOURCE_MANAGEMENT = "resource_management"
    STATE_MANAGEMENT = "state_management"


@traces(spec="035", story="US-006", reqs=["REQ-035-US-006-008"])
@dataclass
class EdgeCase:
    """Definition of an edge case."""

    id: str
    name: str
    description: str
    category: EdgeCaseCategory
    severity: EdgeCaseSeverity
    task_types: list[str]  # Task types this applies to
    detection_patterns: list[str]  # Regex patterns to detect in code
    example_input: str | None = None
    example_output: str | None = None
    mitigation: str | None = None
    created_at: datetime = field(default_factory=datetime.now)
    frequency_score: float = 0.0  # How often this edge case occurs (0-1)


@traces(
    spec="035",
    story="US-006",
    reqs=["REQ-035-US-006-002", "REQ-035-US-006-004"],
)
@dataclass
class EdgeCaseOccurrence:
    """An occurrence of an edge case in a specific task."""

    edge_case_id: str
    task_id: str
    detected_at: datetime = field(default_factory=datetime.now)
    is_tested: bool = False
    test_coverage: float = 0.0  # 0-100%
    code_location: str | None = None


@traces(spec="035", story="US-006", reqs=["REQ-035-US-006-006"])
@dataclass
class EdgeCaseCoverage:
    """Coverage information for edge cases in a task."""

    task_id: str
    total_edge_cases: int = 0
    tested_edge_cases: int = 0
    untested_edge_cases: list[str] = field(default_factory=list)
    coverage_percentage: float = 0.0
    last_updated: datetime = field(default_factory=datetime.now)


@traces(spec="035", story="US-006", reqs=["REQ-035-US-006-009"])
class EdgeCaseDatabase:
    """Database for storing and retrieving edge cases."""

    def __init__(self, storage_path: str | None = None):
        """Initialize the edge case database.

        Args:
            storage_path: Path for persistent storage
        """
        self._edge_cases: dict[str, EdgeCase] = {}
        self._occurrences: dict[str, list[EdgeCaseOccurrence]] = defaultdict(list)
        self._coverage: dict[str, EdgeCaseCoverage] = {}
        self._storage_path = storage_path

        # Initialize with common edge cases
        self._init_common_edge_cases()

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-001"])
    def _init_common_edge_cases(self) -> None:
        """Initialize database with common edge cases."""
        common_cases = [
            # Input Validation
            EdgeCase(
                id="ec-empty-input",
                name="Empty Input Handling",
                description="System should handle empty input gracefully",
                category=EdgeCaseCategory.INPUT_VALIDATION,
                severity=EdgeCaseSeverity.HIGH,
                task_types=["BUILD", "REFACTOR", "TEST"],
                detection_patterns=[r"if.*empty|if.*len\(.*\)\s*==\s*0|if not.*:"],
                example_input="[] or '' or {}",
                example_output="Appropriate error or default value",
                mitigation="Add explicit empty input checks",
            ),
            EdgeCase(
                id="ec-null-none",
                name="Null/None Value Handling",
                description="System should handle null/None values",
                category=EdgeCaseCategory.INPUT_VALIDATION,
                severity=EdgeCaseSeverity.HIGH,
                task_types=["BUILD", "REFACTOR", "DEBUG", "TEST"],
                detection_patterns=[r"if.*is None|if.*is not None|if.*== None"],
                example_input="None",
                example_output="TypeError or handled gracefully",
                mitigation="Add null checks before operations",
            ),
            EdgeCase(
                id="ec-invalid-type",
                name="Invalid Type Handling",
                description="System should validate input types",
                category=EdgeCaseCategory.INPUT_VALIDATION,
                severity=EdgeCaseSeverity.MEDIUM,
                task_types=["BUILD", "REFACTOR", "TEST"],
                detection_patterns=[r"isinstance\(|type\(.*\)\s*==\s*|TypeError"],
                example_input="'string' where int expected",
                example_output="TypeError or validation error",
                mitigation="Use type hints and validation",
            ),
            # Boundary Conditions
            EdgeCase(
                id="ec-max-int",
                name="Integer Overflow",
                description="System should handle maximum integer values",
                category=EdgeCaseCategory.BOUNDARY_CONDITION,
                severity=EdgeCaseSeverity.MEDIUM,
                task_types=["BUILD", "REFACTOR", "TEST"],
                detection_patterns=[r"sys\.maxsize|float\('inf'\)|OverflowError"],
                example_input="sys.maxsize + 1",
                example_output="OverflowError or large number",
                mitigation="Use arbitrary precision integers or bounds checking",
            ),
            EdgeCase(
                id="ec-zero-division",
                name="Division by Zero",
                description="System should handle division by zero",
                category=EdgeCaseCategory.BOUNDARY_CONDITION,
                severity=EdgeCaseSeverity.CRITICAL,
                task_types=["BUILD", "REFACTOR", "DEBUG", "TEST"],
                detection_patterns=[
                    r"ZeroDivisionError|/\s*\w+|//\s*\w+|division|divided by"
                ],
                example_input="x / 0",
                example_output="ZeroDivisionError",
                mitigation="Check denominator before division",
            ),
            EdgeCase(
                id="ec-boundary-values",
                name="Boundary Value Testing",
                description="Test at exact boundary values (min, max, off-by-one)",
                category=EdgeCaseCategory.BOUNDARY_CONDITION,
                severity=EdgeCaseSeverity.HIGH,
                task_types=["BUILD", "TEST"],
                detection_patterns=[r"range\(|\[0\]|\[-1\]|\.append\(|\.pop\("],
                example_input="Index at len-1, len, len+1",
                example_output="IndexError on out of bounds",
                mitigation="Validate indices before access",
            ),
            # Error Handling
            EdgeCase(
                id="ec-exception-handling",
                name="Exception Handling Completeness",
                description="All possible exceptions should be caught",
                category=EdgeCaseCategory.ERROR_HANDLING,
                severity=EdgeCaseSeverity.HIGH,
                task_types=["BUILD", "REFACTOR", "DEBUG"],
                detection_patterns=[r"try:|except:|except Exception|except\s*\w+Error"],
                example_input="Network failure, disk full, etc.",
                example_output="Unhandled exception crash",
                mitigation="Add specific exception handlers",
            ),
            EdgeCase(
                id="ec-resource-cleanup",
                name="Resource Cleanup in Errors",
                description="Resources should be released even on errors",
                category=EdgeCaseCategory.ERROR_HANDLING,
                severity=EdgeCaseSeverity.MEDIUM,
                task_types=["BUILD", "REFACTOR"],
                detection_patterns=[r"finally:|with\s+open|contextlib"],
                example_input="Exception during file operation",
                example_output="Resource leak (open file handles)",
                mitigation="Use try/finally or context managers",
            ),
            # Concurrency
            EdgeCase(
                id="ec-race-condition",
                name="Race Condition",
                description="Concurrent access to shared resources",
                category=EdgeCaseCategory.CONCURRENCY,
                severity=EdgeCaseSeverity.HIGH,
                task_types=["BUILD", "REFACTOR", "DEBUG"],
                detection_patterns=[r"threading\.|asyncio|Lock|Semaphore|async def"],
                example_input="Two threads modify same variable",
                example_output="Inconsistent state",
                mitigation="Use locks or atomic operations",
            ),
            EdgeCase(
                id="ec-deadlock",
                name="Deadlock",
                description="Circular waiting for resources",
                category=EdgeCaseCategory.CONCURRENCY,
                severity=EdgeCaseSeverity.HIGH,
                task_types=["BUILD", "DEBUG"],
                detection_patterns=[r"acquire\(|lock\.|\.wait\("],
                example_input="Thread A holds X, wants Y; Thread B holds Y, wants X",
                example_output="Infinite wait",
                mitigation="Use timeout on locks, consistent lock ordering",
            ),
            # Performance
            EdgeCase(
                id="ec-large-input",
                name="Large Input Performance",
                description="System should handle large inputs efficiently",
                category=EdgeCaseCategory.PERFORMANCE,
                severity=EdgeCaseSeverity.MEDIUM,
                task_types=["BUILD", "REFACTOR", "TEST"],
                detection_patterns=[r"for.*in|while.*:|list comprehension|generator"],
                example_input="List with 1M+ items",
                example_output="Memory error or slow performance",
                mitigation="Use generators, streaming, or pagination",
            ),
            EdgeCase(
                id="ec-infinite-loop",
                name="Infinite Loop",
                description="Loops should have termination conditions",
                category=EdgeCaseCategory.PERFORMANCE,
                severity=EdgeCaseSeverity.CRITICAL,
                task_types=["BUILD", "DEBUG"],
                detection_patterns=[r"while True|while\s+\w+\s*:\s*\n", r"recursion"],
                example_input="while True without break",
                example_output="Infinite execution",
                mitigation="Add loop counters, timeouts, or max iterations",
            ),
            # Security
            EdgeCase(
                id="ec-injection",
                name="Injection Attack",
                description="Prevent injection attacks (SQL, command, etc.)",
                category=EdgeCaseCategory.SECURITY,
                severity=EdgeCaseSeverity.CRITICAL,
                task_types=["BUILD", "REVIEW"],
                detection_patterns=[
                    r"execute\(|eval\(|exec\(|subprocess\.call|shell=True"
                ],
                example_input="'; DROP TABLE users; --",
                example_output="SQL injection vulnerability",
                mitigation="Use parameterized queries, avoid eval/exec",
            ),
            EdgeCase(
                id="ec-path-traversal",
                name="Path Traversal",
                description="Prevent access to files outside intended directory",
                category=EdgeCaseCategory.SECURITY,
                severity=EdgeCaseSeverity.HIGH,
                task_types=["BUILD", "REVIEW", "SECURITY"],
                detection_patterns=[r"open\(|\.read\(|\.write\(|pathlib|os\.path"],
                example_input="../../../etc/passwd",
                example_output="Unauthorized file access",
                mitigation="Validate and sanitize paths, use allowlists",
            ),
            # Resource Management
            EdgeCase(
                id="ec-memory-leak",
                name="Memory Leak",
                description="Unused memory not being released",
                category=EdgeCaseCategory.RESOURCE_MANAGEMENT,
                severity=EdgeCaseSeverity.MEDIUM,
                task_types=["BUILD", "DEBUG", "REVIEW"],
                detection_patterns=[r"\.append\(|global|cache\[|lru_cache"],
                example_input="Growing cache without eviction",
                example_output="Out of memory error",
                mitigation="Use weak references, implement cache eviction",
            ),
            # State Management
            EdgeCase(
                id="ec-invalid-state",
                name="Invalid State Transition",
                description="State machine should handle invalid transitions",
                category=EdgeCaseCategory.STATE_MANAGEMENT,
                severity=EdgeCaseSeverity.MEDIUM,
                task_types=["BUILD", "REFACTOR"],
                detection_patterns=[r"state|status|transition|\.set_|enum"],
                example_input="Close already closed connection",
                example_output="InvalidStateError or unexpected behavior",
                mitigation="Validate state before operations",
            ),
        ]

        for case in common_cases:
            self._edge_cases[case.id] = case

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-008"])
    def add_edge_case(self, edge_case: EdgeCase) -> None:
        """Add a new edge case to the database."""
        self._edge_cases[edge_case.id] = edge_case

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-014"])
    def get_edge_case(self, edge_case_id: str) -> EdgeCase | None:
        """Get an edge case by ID."""
        return self._edge_cases.get(edge_case_id)

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-001"])
    def get_edge_cases_for_task(
        self,
        task_type: str,
        category: EdgeCaseCategory | None = None,
        min_severity: EdgeCaseSeverity | None = None,
    ) -> list[EdgeCase]:
        """Get edge cases applicable to a task type."""
        results = []
        for case in self._edge_cases.values():
            if task_type.upper() in [t.upper() for t in case.task_types]:
                if category and case.category != category:
                    continue
                if min_severity:
                    severity_order = ["low", "medium", "high", "critical"]
                    if severity_order.index(case.severity.value) < severity_order.index(
                        min_severity.value
                    ):
                        continue
                results.append(case)
        return results

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-010"])
    def record_occurrence(self, occurrence: EdgeCaseOccurrence) -> None:
        """Record an edge case occurrence."""
        self._occurrences[occurrence.task_id].append(occurrence)

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-014"])
    def get_occurrences(self, task_id: str) -> list[EdgeCaseOccurrence]:
        """Get all occurrences for a task."""
        return self._occurrences.get(task_id, [])

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-006"])
    def get_coverage(self, task_id: str) -> EdgeCaseCoverage:
        """Get edge case coverage for a task."""
        return self._coverage.get(task_id, EdgeCaseCoverage(task_id=task_id))


@traces(
    spec="035",
    story="US-006",
    reqs=["REQ-035-US-006-001", "REQ-035-US-006-010"],
)
class EdgeCaseDetector:
    """Detector for identifying edge cases in code."""

    def __init__(self, database: EdgeCaseDatabase | None = None):
        """Initialize the detector.

        Args:
            database: Edge case database (optional for result-based detection)
        """
        self.database = database if database is not None else EdgeCaseDatabase()

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-010"])
    def detect(self, results: list[VerificationResult]) -> list[TypesEdgeCase]:
        """Detect edge cases from verification results.

        PARTIAL results produce medium-severity edge cases.
        ERROR results produce high-severity edge cases.
        SATISFIED and NOT_MET do not produce edge cases.

        Args:
            results: List of verification results

        Returns:
            List of EdgeCase instances from types module
        """
        cases: list[TypesEdgeCase] = []
        for result in results:
            if result.status == VerificationStatus.PARTIAL:
                cases.append(
                    TypesEdgeCase(
                        constraint_id=result.constraint_id,
                        description=result.details or "Partial verification",
                        severity="medium",
                    )
                )
            elif result.status == VerificationStatus.ERROR:
                cases.append(
                    TypesEdgeCase(
                        constraint_id=result.constraint_id,
                        description=result.details or "Verification error",
                        severity="high",
                    )
                )
        return cases

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-010"])
    def detect_boundary_cases(
        self, result: VerificationResult, tolerance: float = 0.05
    ) -> list[TypesEdgeCase]:
        """Detect boundary/near-miss cases from a verification result.

        If actual is within tolerance of expected, it is a near-boundary case.

        Args:
            result: A single verification result with expected/actual values
            tolerance: Fractional tolerance (e.g. 0.05 = 5%)

        Returns:
            List of EdgeCase instances (0 or 1)
        """
        cases: list[TypesEdgeCase] = []
        if (
            result.expected is not None
            and result.actual is not None
            and isinstance(result.expected, (int, float))
            and isinstance(result.actual, (int, float))
            and result.expected != 0
        ):
            diff = abs(result.expected - result.actual) / abs(result.expected)
            if 0 < diff <= tolerance:
                cases.append(
                    TypesEdgeCase(
                        constraint_id=result.constraint_id,
                        description=f"Near boundary: expected={result.expected}, actual={result.actual}",
                        severity="low",
                    )
                )
        return cases

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-010"])
    def detect_edge_cases(
        self, code: str, task_type: str, task_id: str
    ) -> list[tuple[EdgeCase, list[int]]]:
        """Detect edge cases in code.

        Args:
            code: Source code to analyze
            task_type: Type of task (BUILD, REFACTOR, etc.)
            task_id: Unique task identifier

        Returns:
            List of (edge_case, line_numbers) tuples
        """
        detected = []
        lines = code.split("\n")

        # Get applicable edge cases
        edge_cases = self.database.get_edge_cases_for_task(task_type)

        for edge_case in edge_cases:
            matching_lines = []

            for i, line in enumerate(lines, 1):
                for pattern in edge_case.detection_patterns:
                    try:
                        if re.search(pattern, line, re.IGNORECASE):
                            matching_lines.append(i)
                            break
                    except re.error:
                        continue

            if matching_lines:
                detected.append((edge_case, matching_lines))

                # Record occurrence
                occurrence = EdgeCaseOccurrence(
                    edge_case_id=edge_case.id,
                    task_id=task_id,
                    code_location=f"lines {min(matching_lines)}-{max(matching_lines)}",
                )
                self.database.record_occurrence(occurrence)

        return detected

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-003"])
    def suggest_edge_cases(
        self, code: str, task_type: str, existing_tests: str | None = None
    ) -> list[EdgeCase]:
        """Suggest edge cases that should be tested.

        Args:
            code: Source code
            task_type: Type of task
            existing_tests: Optional existing test code

        Returns:
            List of suggested edge cases
        """
        detected = self.detect_edge_cases(code, task_type, "suggestion")

        suggestions = []
        for edge_case, _ in detected:
            # Check if already tested
            if existing_tests and self._is_edge_case_tested(edge_case, existing_tests):
                continue
            suggestions.append(edge_case)

        # Sort by severity (critical first)
        severity_order = ["critical", "high", "medium", "low"]
        suggestions.sort(key=lambda e: severity_order.index(e.severity.value))

        return suggestions

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-002"])
    def _is_edge_case_tested(self, edge_case: EdgeCase, test_code: str) -> bool:
        """Check if an edge case is already tested."""
        # Check if detection patterns appear in tests
        for pattern in edge_case.detection_patterns:
            try:
                if re.search(pattern, test_code, re.IGNORECASE):
                    return True
            except re.error:
                continue
        # Fallback: check if example_input values appear in test code
        if edge_case.example_input:
            for fragment in edge_case.example_input.split(" or "):
                fragment = fragment.strip().strip("'\"")
                if fragment and fragment in test_code:
                    return True
        # Fallback: check if edge case name keywords appear in test code
        name_words = [w.lower() for w in edge_case.name.split() if len(w) > 3]
        test_lower = test_code.lower()
        if any(w in test_lower for w in name_words):
            return True
        return False

    @traces(
        spec="035", story="US-006", reqs=["REQ-035-US-006-006", "REQ-035-US-006-012"]
    )
    def calculate_coverage(
        self, code: str, test_code: str, task_id: str, task_type: str
    ) -> EdgeCaseCoverage:
        """Calculate edge case coverage for a task."""
        # Detect edge cases in code
        detected = self.detect_edge_cases(code, task_type, task_id)

        # Check which are tested
        tested = []
        untested = []

        for edge_case, _ in detected:
            if self._is_edge_case_tested(edge_case, test_code):
                tested.append(edge_case.id)
            else:
                untested.append(edge_case.id)

        total = len(detected)
        coverage = EdgeCaseCoverage(
            task_id=task_id,
            total_edge_cases=total,
            tested_edge_cases=len(tested),
            untested_edge_cases=untested,
            coverage_percentage=(len(tested) / total * 100) if total > 0 else 0.0,
        )

        self.database._coverage[task_id] = coverage
        return coverage

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-007"])
    def generate_test_suggestions(
        self, edge_case: EdgeCase, function_signature: str
    ) -> str:
        """Generate test code suggestion for an edge case."""
        suggestions = {
            EdgeCaseCategory.INPUT_VALIDATION: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    # Test with invalid input: {edge_case.example_input}
    with pytest.raises((ValueError, TypeError)):
        {function_signature}({edge_case.example_input or "None"})
""",
            EdgeCaseCategory.BOUNDARY_CONDITION: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    # Test boundary values
    result = {function_signature}({edge_case.example_input or "boundary_value"})
    assert result is not None  # Add specific assertions
""",
            EdgeCaseCategory.ERROR_HANDLING: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    # Test error handling
    with pytest.raises(Exception):
        {function_signature}({edge_case.example_input or "error_condition"})
    # Verify cleanup occurred (if applicable)
""",
            EdgeCaseCategory.CONCURRENCY: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    import concurrent.futures
    # Test concurrent access
    with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
        futures = [executor.submit({function_signature}) for _ in range(2)]
        results = [f.result() for f in futures]
    # Verify consistent results
""",
            EdgeCaseCategory.PERFORMANCE: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    import time
    start = time.time()
    result = {function_signature}({edge_case.example_input or "large_input"})
    duration = time.time() - start
    assert duration < 1.0  # Should complete in less than 1 second
""",
            EdgeCaseCategory.SECURITY: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    # Test security vulnerability
    malicious_input = {repr(edge_case.example_input) if edge_case.example_input else "'malicious_input'"}
    with pytest.raises((ValueError, SecurityError)):
        {function_signature}(malicious_input)
""",
            EdgeCaseCategory.RESOURCE_MANAGEMENT: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    import resource
    initial_memory = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    # Run operation multiple times
    for _ in range(100):
        {function_signature}()
    final_memory = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
    # Memory should not grow significantly
    assert final_memory - initial_memory < 1024 * 1024  # 1MB threshold
""",
            EdgeCaseCategory.STATE_MANAGEMENT: f"""
# Test: {edge_case.name}
def test_{edge_case.id.replace("-", "_")}(self):
    \"\"\"{edge_case.description}\"\"\"
    # Test invalid state transition
    obj = create_object()
    obj.close()
    with pytest.raises((InvalidStateError, RuntimeError)):
        obj.close()  # Should not allow double-close
""",
        }

        return suggestions.get(
            edge_case.category,
            f"# Test: {edge_case.name}\ndef test_{edge_case.id.replace('-', '_')}(self):\n    pass",
        )


@traces(spec="035", story="US-006", reqs=["REQ-035-US-006-014"])
class EdgeCaseReporter:
    """Reporter for edge case analysis results."""

    def __init__(self, database: EdgeCaseDatabase):
        """Initialize the reporter.

        Args:
            database: Edge case database
        """
        self.database = database

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-014"])
    def generate_report(self, task_id: str) -> dict[str, Any]:
        """Generate edge case report for a task."""
        occurrences = self.database.get_occurrences(task_id)
        coverage = self.database.get_coverage(task_id)

        # Group by category
        by_category = defaultdict(list)
        for occ in occurrences:
            edge_case = self.database.get_edge_case(occ.edge_case_id)
            if edge_case:
                by_category[edge_case.category.value].append(
                    {
                        "edge_case_id": occ.edge_case_id,
                        "edge_case_name": edge_case.name,
                        "severity": edge_case.severity.value,
                        "is_tested": occ.is_tested,
                        "code_location": occ.code_location,
                    }
                )

        return {
            "task_id": task_id,
            "generated_at": datetime.now().isoformat(),
            "summary": {
                "total_edge_cases_detected": len(occurrences),
                "tested_count": coverage.tested_edge_cases,
                "untested_count": len(coverage.untested_edge_cases),
                "coverage_percentage": coverage.coverage_percentage,
            },
            "by_category": dict(by_category),
            "untested_edge_cases": coverage.untested_edge_cases,
            "recommendations": self._generate_recommendations(occurrences, coverage),
        }

    @traces(spec="035", story="US-006", reqs=["REQ-035-US-006-014"])
    def _generate_recommendations(
        self, occurrences: list[EdgeCaseOccurrence], coverage: EdgeCaseCoverage
    ) -> list[str]:
        """Generate recommendations based on analysis."""
        recommendations = []

        if coverage.coverage_percentage < 50:
            recommendations.append(
                f"CRITICAL: Edge case coverage is only {coverage.coverage_percentage:.1f}%. "
                "Add tests for the untested edge cases."
            )
        elif coverage.coverage_percentage < 80:
            recommendations.append(
                f"WARNING: Edge case coverage is {coverage.coverage_percentage:.1f}%. "
                "Consider adding more edge case tests."
            )

        # Check for critical untested edge cases
        critical_untested = []
        for edge_case_id in coverage.untested_edge_cases:
            edge_case = self.database.get_edge_case(edge_case_id)
            if edge_case and edge_case.severity == EdgeCaseSeverity.CRITICAL:
                critical_untested.append(edge_case.name)

        if critical_untested:
            recommendations.append(
                f"URGENT: Untested critical edge cases: {', '.join(critical_untested)}"
            )

        return recommendations

    def export_json(self, task_id: str) -> str:
        """Export report as JSON."""
        report = self.generate_report(task_id)
        return json.dumps(report, indent=2)

    def export_markdown(self, task_id: str) -> str:
        """Export report as Markdown."""
        report = self.generate_report(task_id)

        lines = [
            f"# Edge Case Analysis Report: {task_id}",
            "",
            f"**Generated:** {report['generated_at']}",
            "",
            "## Summary",
            "",
            f"- **Total Edge Cases Detected:** {report['summary']['total_edge_cases_detected']}",
            f"- **Tested:** {report['summary']['tested_count']}",
            f"- **Untested:** {report['summary']['untested_count']}",
            f"- **Coverage:** {report['summary']['coverage_percentage']:.1f}%",
            "",
            "## Recommendations",
            "",
        ]

        for rec in report["recommendations"]:
            lines.append(f"- {rec}")

        lines.extend(["", "## By Category", ""])

        for category, cases in report["by_category"].items():
            lines.append(f"### {category.replace('_', ' ').title()}")
            lines.append("")
            for case in cases:
                status = "✅ Tested" if case["is_tested"] else "❌ Untested"
                lines.append(
                    f"- **{case['edge_case_name']}** ({case['severity']}) - {status}"
                )
            lines.append("")

        return "\n".join(lines)