scripts-solves-edge-builder

#!/usr/bin/env python3 """ CP-23: SOLVES Edge Builder (ADR-151)

Creates SOLVES edges for ErrorSolution nodes.

Edge: error_solution:X -> error_solution:X (self-referential pattern) OR error_solution:X -> virtual error pattern node Source: org.db error_solutions table Properties: error_pattern, solution_count, success_rate

Error solutions represent patterns (error_type + signature) that are solved. Since errors are not separate nodes, we create self-referential edges with the error pattern encoded in properties.

Created: 2026-02-03 Track: J (Memory Intelligence) Task: J.3.5.3 """

import json import logging from pathlib import Path from typing import Any, Dict, Generator, Optional, Tuple

from .base_edge_builder import SQLiteSourceEdgeBuilder

logger = logging.getLogger(name)

class SolvesEdgeBuilder(SQLiteSourceEdgeBuilder): """ Build SOLVES edges for error solutions.

Since error patterns are not separate nodes in kg_nodes (they're
embedded in error_solution nodes), we create edges that capture
the solving relationship with the pattern in properties.

Approach: Create edges from error_solution nodes to themselves
with the error pattern stored in edge properties, or create
inter-solution edges when multiple solutions exist for similar patterns.
"""

@property
def edge_type(self) -> str:
    return "SOLVES"

def extract_edges(self) -> Generator[Tuple[str, str, Dict[str, Any]], None, None]:
    """
    Extract SOLVES edges from error_solutions table.

    Creates edges representing what error patterns each solution solves.
    Also links solutions that solve similar error patterns.

    Yields:
        Tuple of (from_node_id, to_node_id, properties)
    """
    source_conn = self.connect_source()

    # Group solutions by error pattern for cross-linking
    pattern_to_solutions: Dict[str, list] = {}

    try:
        cursor = source_conn.execute("""
            SELECT
                id,
                error_hash,
                error_type,
                error_signature,
                error_context,
                success_count,
                failure_count,
                language
            FROM error_solutions
            ORDER BY error_type, error_signature
        """)

        solutions = list(cursor)

        # First pass: group by error pattern
        for row in solutions:
            error_type = row['error_type'] or 'unknown'
            error_signature = row['error_signature'] or ''
            error_hash = row['error_hash'] or str(row['id'])

            # Create pattern key (normalized)
            pattern_key = f"{error_type}:{error_signature[:100]}"

            if pattern_key not in pattern_to_solutions:
                pattern_to_solutions[pattern_key] = []

            pattern_to_solutions[pattern_key].append({
                'id': row['id'],
                'error_hash': error_hash,  # Use error_hash for node_id matching
                'error_type': error_type,
                'error_signature': error_signature,
                'success_count': row['success_count'] or 0,
                'failure_count': row['failure_count'] or 0,
                'language': row['language'],
            })

        # Second pass: create edges
        for pattern_key, solutions_list in pattern_to_solutions.items():
            # Create self-referential edges to capture the SOLVES relationship
            for solution in solutions_list:
                # Use error_hash for node_id (matches error_solution_extractor)
                error_hash = solution['error_hash']
                node_id = f"error_solution:{error_hash}"

                # Calculate success rate
                total = solution['success_count'] + solution['failure_count']
                success_rate = solution['success_count'] / total if total > 0 else 0.0

                # Self-referential edge with pattern info
                # This captures "this solution solves this error pattern"
                properties = {
                    'error_pattern': pattern_key,
                    'error_type': solution['error_type'],
                    'solution_count': 1,
                    'success_rate': round(success_rate, 3),
                    'total_applications': total,
                }

                if solution['language']:
                    properties['language'] = solution['language']

                # Create self-edge
                yield (node_id, node_id, properties)

            # If multiple solutions for same pattern, link them
            if len(solutions_list) > 1:
                # Sort by success rate (descending)
                sorted_solutions = sorted(
                    solutions_list,
                    key=lambda s: s['success_count'] / max(s['success_count'] + s['failure_count'], 1),
                    reverse=True
                )

                # Link solutions for same pattern (best solution → alternatives)
                best = sorted_solutions[0]
                best_node = f"error_solution:{best['error_hash']}"

                for alt in sorted_solutions[1:]:
                    alt_node = f"error_solution:{alt['error_hash']}"

                    properties = {
                        'relationship': 'alternative_solution',
                        'error_pattern': pattern_key,
                        'relative_ranking': sorted_solutions.index(alt) + 1,
                    }

                    yield (best_node, alt_node, properties)

    except Exception as e:
        logger.error(f"Error extracting SOLVES edges: {e}")
        return