stack/src/sem/plan.py

"""Layer 4 — Planner: prioritise uncovered nodes and suggest test targets.

Implements the priority heuristic from the design doc::

    priority =
        uncovered_branch * 5
      + ty_related * 4
      + ruff_warning * 2
      + inside_partially_tested_function * 3
      - prior_failures * 2
      - likely_unreachable * 5
"""

from __future__ import annotations

from sem.nodes import NodeKind, SemanticNode

# Branch-like node kinds that carry higher weight when uncovered.
_BRANCH_KINDS = {
    NodeKind.BRANCH_IF,
    NodeKind.BRANCH_ELSE,
    NodeKind.BRANCH_ELIF,
    NodeKind.MATCH_CASE,
}

_EXCEPTION_KINDS = {
    NodeKind.EXCEPT_HANDLER,
}


def _is_branch(node: SemanticNode) -> bool:
    return node.kind in _BRANCH_KINDS


def _is_exception_path(node: SemanticNode) -> bool:
    return node.kind in _EXCEPTION_KINDS


def _function_coverage_ratio(
    node: SemanticNode, all_nodes: list[SemanticNode]
) -> float:
    """Return fraction of nodes hit in the same function context."""
    if not node.symbol_context:
        return 0.0
    fn_ctx = node.symbol_context[0] if node.symbol_context else ""
    siblings = [
        n
        for n in all_nodes
        if n.module == node.module
        and n.symbol_context
        and n.symbol_context[0] == fn_ctx
    ]
    if not siblings:
        return 0.0
    hit = sum(1 for s in siblings if s.runtime.hit)
    return hit / len(siblings)


def score_node(node: SemanticNode, all_nodes: list[SemanticNode]) -> float:
    """Compute priority score for a single node."""
    score = 0.0

    # Uncovered branch
    if not node.runtime.hit and _is_branch(node):
        score += 5.0

    # ty diagnostics present
    if node.ty_codes:
        score += 4.0

    # Ruff warnings present
    if node.ruff_codes:
        score += 2.0

    # Inside a partially tested function (some siblings hit, this one not)
    fn_ratio = _function_coverage_ratio(node, all_nodes)
    if 0.0 < fn_ratio < 1.0 and not node.runtime.hit:
        score += 3.0

    # Exception paths — valuable targets
    if not node.runtime.hit and _is_exception_path(node):
        score += 3.0

    # Penalise prior failures
    score -= node.attempts * 2.0

    # Penalise already-covered nodes
    if node.runtime.hit:
        score -= 5.0

    return score


def rank_nodes(nodes: list[SemanticNode]) -> list[SemanticNode]:
    """Score and sort nodes by descending priority.

    Mutates ``node.priority`` in place, then returns nodes sorted
    highest-priority-first.
    """
    for node in nodes:
        node.priority = score_node(node, nodes)
    return sorted(nodes, key=lambda n: n.priority, reverse=True)


def next_targets(nodes: list[SemanticNode], *, limit: int = 10) -> list[SemanticNode]:
    """Return the top-*limit* uncovered nodes worth testing next."""
    ranked = rank_nodes(nodes)
    return [n for n in ranked if not n.runtime.hit and n.priority > 0][:limit]