feat: sem — semantic coverage orchestration (AST + Ruff + ty + coverage)

4-layer architecture for tracking coverage as semantic AST nodes rather
than raw line numbers: parse → enrich → runtime → plan.  Stable node
identities, Ruff/ty diagnostic attachment, coverage mapping, priority
scoring heuristic, and drift-resistant rolling state persistence.

39 tests, ruff clean.

pyproject.toml

@@ -87,6 +87,9 @@ perf = [
     "opentelemetry-semantic-conventions>=0.46b0",
     "psutil>=5.9.0",
 ]
+sem = [
+    "coverage>=7.13.4",
+]
 lake = [
     "stack[aco]",
     "databricks-sdk>=0.85.0",
@@ -153,6 +156,6 @@ ignore = ["E501", "E741"]
 testpaths = ["tests"]
 
 [tool.uv.build-backend]
-module-name = ["aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "conf", "opps", "perf", "pfs", "rex"]
+module-name = ["aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "conf", "opps", "perf", "pfs", "rex", "sem"]
 namespace = true
 source-exclude = ["compose.yml","infra/**","data/**","notebooks/**","tuva/**","assets/**","docs/**","dev/**","bundle/**","cloud/**"]

src/sem/__init__.py

@@ -0,0 +1,30 @@
+"""Semantic coverage orchestration.
+
+AST-based coverage tracking that attaches Ruff diagnostics, ty type
+signals, and runtime coverage data to semantic nodes (functions, branches,
+exception handlers) rather than raw line numbers.
+
+Four layers:
+    1. parse   — build AST index of semantic nodes
+    2. enrich  — attach Ruff + ty diagnostics
+    3. runtime — attach coverage / per-test execution data
+    4. plan    — prioritise uncovered nodes, suggest test targets
+"""
+
+from sem.enrich import attach_ruff, attach_ty
+from sem.nodes import NodeKind, SemanticNode
+from sem.parse import parse_module, parse_tree
+from sem.plan import next_targets, rank_nodes
+from sem.runtime import attach_coverage
+
+__all__ = [
+    "NodeKind",
+    "SemanticNode",
+    "parse_module",
+    "parse_tree",
+    "attach_ruff",
+    "attach_ty",
+    "attach_coverage",
+    "rank_nodes",
+    "next_targets",
+]

src/sem/enrich.py

@@ -0,0 +1,97 @@
+"""Layer 2 — Enrichment: attach Ruff and ty diagnostics to semantic nodes.
+
+Runs ``ruff check`` and ``ty check`` as subprocesses, parses their JSON
+output, and maps each diagnostic to the semantic node whose span
+contains the reported line.
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+from sem.nodes import SemanticNode
+
+
+def _run_tool(cmd: list[str]) -> list[dict]:
+    """Run a linter/checker and return parsed JSON diagnostics."""
+    try:
+        result = subprocess.run(
+            cmd,
+            capture_output=True,
+            text=True,
+            timeout=120,
+        )
+    except FileNotFoundError:
+        return []
+    except subprocess.TimeoutExpired:
+        return []
+
+    raw = result.stdout.strip()
+    if not raw:
+        return []
+
+    try:
+        return json.loads(raw)
+    except json.JSONDecodeError:
+        return []
+
+
+def _find_node(nodes: list[SemanticNode], line: int) -> SemanticNode | None:
+    """Find the tightest (smallest span) node containing *line*."""
+    best: SemanticNode | None = None
+    best_size = float("inf")
+    for node in nodes:
+        if node.span.start_line <= line <= node.span.end_line:
+            size = node.span.end_line - node.span.start_line
+            if size < best_size:
+                best = node
+                best_size = size
+    return best
+
+
+def attach_ruff(
+    nodes: list[SemanticNode],
+    path: Path,
+    *,
+    ruff_bin: str = "ruff",
+) -> list[SemanticNode]:
+    """Run ``ruff check --output-format=json`` and attach codes to nodes.
+
+    Returns the same list (mutated in place) for chaining convenience.
+    """
+    diagnostics = _run_tool(
+        [ruff_bin, "check", "--output-format=json", "--no-fix", str(path)]
+    )
+    for diag in diagnostics:
+        line = diag.get("location", {}).get("row", 0)
+        code = diag.get("code", "")
+        if not line or not code:
+            continue
+        target = _find_node(nodes, line)
+        if target and code not in target.ruff_codes:
+            target.ruff_codes.append(code)
+    return nodes
+
+
+def attach_ty(
+    nodes: list[SemanticNode],
+    path: Path,
+    *,
+    ty_bin: str = "ty",
+) -> list[SemanticNode]:
+    """Run ``ty check --output-format=json`` and attach codes to nodes.
+
+    Returns the same list (mutated in place) for chaining convenience.
+    """
+    diagnostics = _run_tool([ty_bin, "check", "--output-format=json", str(path)])
+    for diag in diagnostics:
+        line = diag.get("location", {}).get("row", 0)
+        code = diag.get("code", diag.get("rule", ""))
+        if not line or not code:
+            continue
+        target = _find_node(nodes, line)
+        if target and code not in target.ty_codes:
+            target.ty_codes.append(code)
+    return nodes

src/sem/nodes.py

@@ -0,0 +1,119 @@
+"""Semantic node model.
+
+Each meaningful AST construct becomes a ``SemanticNode`` with a stable
+identity derived from its module path, qualified symbol name, node kind,
+and ordinal position — never from line numbers.
+"""
+
+from __future__ import annotations
+
+import enum
+import hashlib
+from dataclasses import dataclass, field
+
+
+class NodeKind(enum.Enum):
+    """Taxonomy of tracked AST constructs."""
+
+    FUNCTION = "function"
+    METHOD = "method"
+    CLASS = "class"
+    BRANCH_IF = "branch_if"
+    BRANCH_ELSE = "branch_else"
+    BRANCH_ELIF = "branch_elif"
+    MATCH_CASE = "match_case"
+    EXCEPT_HANDLER = "except_handler"
+    FOR_LOOP = "for_loop"
+    WHILE_LOOP = "while_loop"
+    WITH_BLOCK = "with_block"
+    RETURN = "return"
+    RAISE = "raise"
+    ASSERT = "assert"
+
+
+def _make_node_id(
+    module: str,
+    symbol_context: list[str],
+    kind: NodeKind,
+    ordinal: int,
+) -> str:
+    """Build a stable, human-readable node identity string.
+
+    Example: ``app.config::load_settings::branch_if[1]``
+    """
+    parts = [module]
+    if symbol_context:
+        parts.append("::".join(symbol_context))
+    parts.append(f"{kind.value}[{ordinal}]")
+    return "::".join(parts)
+
+
+@dataclass
+class Span:
+    """Byte-offset span within a source file (line numbers as metadata only)."""
+
+    start_line: int
+    end_line: int
+    start_col: int = 0
+    end_col: int = 0
+
+
+@dataclass
+class RuntimeInfo:
+    """Runtime coverage data attached after test execution."""
+
+    hit: bool = False
+    hit_count: int = 0
+    tests_seen: list[str] = field(default_factory=list)
+
+
+@dataclass
+class SemanticNode:
+    """A single tracked semantic unit in the codebase."""
+
+    node_id: str
+    kind: NodeKind
+    span: Span
+    module: str
+    symbol_context: list[str] = field(default_factory=list)
+    source_hash: str = ""
+
+    # Enrichment: Ruff diagnostics
+    ruff_codes: list[str] = field(default_factory=list)
+
+    # Enrichment: ty diagnostics
+    ty_codes: list[str] = field(default_factory=list)
+
+    # Runtime coverage
+    runtime: RuntimeInfo = field(default_factory=RuntimeInfo)
+
+    # Planner state
+    status: str = "unexplored"
+    priority: float = 0.0
+    attempts: int = 0
+    intent_tags: list[str] = field(default_factory=list)
+
+    @staticmethod
+    def build(
+        module: str,
+        symbol_context: list[str],
+        kind: NodeKind,
+        ordinal: int,
+        span: Span,
+        source_fragment: str = "",
+    ) -> SemanticNode:
+        """Construct a node with a stable ID and optional source hash."""
+        node_id = _make_node_id(module, symbol_context, kind, ordinal)
+        src_hash = ""
+        if source_fragment:
+            src_hash = hashlib.sha1(
+                source_fragment.encode(), usedforsecurity=False
+            ).hexdigest()[:12]
+        return SemanticNode(
+            node_id=node_id,
+            kind=kind,
+            span=span,
+            module=module,
+            symbol_context=list(symbol_context),
+            source_hash=src_hash,
+        )

src/sem/parse.py

@@ -0,0 +1,168 @@
+"""Layer 1 — Parse: build semantic nodes from Python AST.
+
+Walks a module's AST and emits a ``SemanticNode`` for every tracked
+construct: functions, methods, classes, branches, loops, exception
+handlers, returns, and raises.
+"""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+from sem.nodes import NodeKind, SemanticNode, Span
+
+
+def _span(node: ast.AST) -> Span:
+    return Span(
+        start_line=getattr(node, "lineno", 0),
+        end_line=getattr(node, "end_lineno", 0) or getattr(node, "lineno", 0),
+        start_col=getattr(node, "col_offset", 0),
+        end_col=getattr(node, "end_col_offset", 0),
+    )
+
+
+def _source_lines(source: str, span: Span) -> str:
+    lines = source.splitlines()
+    start = max(0, span.start_line - 1)
+    end = min(len(lines), span.end_line)
+    return "\n".join(lines[start:end])
+
+
+class _Collector(ast.NodeVisitor):
+    """Walk AST and collect semantic nodes with stable ordinal counters."""
+
+    def __init__(self, module: str, source: str) -> None:
+        self.module = module
+        self.source = source
+        self.nodes: list[SemanticNode] = []
+        self._symbol_stack: list[str] = []
+        self._class_depth: int = 0
+        self._counters: dict[str, int] = {}
+
+    def _next_ordinal(self, kind: NodeKind) -> int:
+        ctx = "::".join([self.module] + self._symbol_stack)
+        key = f"{ctx}::{kind.value}"
+        n = self._counters.get(key, 0)
+        self._counters[key] = n + 1
+        return n
+
+    def _emit(self, kind: NodeKind, node: ast.AST) -> None:
+        sp = _span(node)
+        ordinal = self._next_ordinal(kind)
+        frag = _source_lines(self.source, sp)
+        self.nodes.append(
+            SemanticNode.build(
+                module=self.module,
+                symbol_context=list(self._symbol_stack),
+                kind=kind,
+                ordinal=ordinal,
+                span=sp,
+                source_fragment=frag,
+            )
+        )
+
+    # -- functions / methods --------------------------------------------------
+
+    def _visit_func(self, node: ast.FunctionDef | ast.AsyncFunctionDef) -> None:
+        kind = NodeKind.METHOD if self._class_depth > 0 else NodeKind.FUNCTION
+        self._symbol_stack.append(node.name)
+        self._emit(kind, node)
+        self.generic_visit(node)
+        self._symbol_stack.pop()
+
+    def visit_FunctionDef(self, node: ast.FunctionDef) -> None:
+        self._visit_func(node)
+
+    def visit_AsyncFunctionDef(self, node: ast.AsyncFunctionDef) -> None:
+        self._visit_func(node)
+
+    # -- classes --------------------------------------------------------------
+
+    def visit_ClassDef(self, node: ast.ClassDef) -> None:
+        self._symbol_stack.append(node.name)
+        self._emit(NodeKind.CLASS, node)
+        self._class_depth += 1
+        self.generic_visit(node)
+        self._class_depth -= 1
+        self._symbol_stack.pop()
+
+    # -- control flow ---------------------------------------------------------
+
+    def visit_If(self, node: ast.If) -> None:
+        self._emit(NodeKind.BRANCH_IF, node)
+        if node.orelse:
+            for alt in node.orelse:
+                if isinstance(alt, ast.If):
+                    self._emit(NodeKind.BRANCH_ELIF, alt)
+                else:
+                    self._emit(NodeKind.BRANCH_ELSE, node)
+                    break
+        self.generic_visit(node)
+
+    def visit_Match(self, node: ast.Match) -> None:
+        for case in node.cases:
+            self._emit(NodeKind.MATCH_CASE, case)
+        self.generic_visit(node)
+
+    def visit_For(self, node: ast.For) -> None:
+        self._emit(NodeKind.FOR_LOOP, node)
+        self.generic_visit(node)
+
+    def visit_AsyncFor(self, node: ast.AsyncFor) -> None:
+        self._emit(NodeKind.FOR_LOOP, node)
+        self.generic_visit(node)
+
+    def visit_While(self, node: ast.While) -> None:
+        self._emit(NodeKind.WHILE_LOOP, node)
+        self.generic_visit(node)
+
+    def visit_With(self, node: ast.With) -> None:
+        self._emit(NodeKind.WITH_BLOCK, node)
+        self.generic_visit(node)
+
+    def visit_AsyncWith(self, node: ast.AsyncWith) -> None:
+        self._emit(NodeKind.WITH_BLOCK, node)
+        self.generic_visit(node)
+
+    # -- exception handling ---------------------------------------------------
+
+    def visit_ExceptHandler(self, node: ast.ExceptHandler) -> None:
+        self._emit(NodeKind.EXCEPT_HANDLER, node)
+        self.generic_visit(node)
+
+    # -- returns / raises -----------------------------------------------------
+
+    def visit_Return(self, node: ast.Return) -> None:
+        self._emit(NodeKind.RETURN, node)
+
+    def visit_Raise(self, node: ast.Raise) -> None:
+        self._emit(NodeKind.RAISE, node)
+
+    # -- assert ---------------------------------------------------------------
+
+    def visit_Assert(self, node: ast.Assert) -> None:
+        self._emit(NodeKind.ASSERT, node)
+
+
+def parse_module(path: Path, module: str | None = None) -> list[SemanticNode]:
+    """Parse a single Python file and return its semantic nodes.
+
+    Parameters
+    ----------
+    path
+        Path to the ``.py`` file.
+    module
+        Dotted module name.  If *None*, derived from the file stem.
+    """
+    source = path.read_text(encoding="utf-8")
+    tree = ast.parse(source, filename=str(path))
+    mod_name = module or path.stem
+    return parse_tree(tree, mod_name, source)
+
+
+def parse_tree(tree: ast.Module, module: str, source: str = "") -> list[SemanticNode]:
+    """Parse an already-built AST and return semantic nodes."""
+    collector = _Collector(module, source)
+    collector.visit(tree)
+    return collector.nodes

src/sem/plan.py

@@ -0,0 +1,108 @@
+"""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]

src/sem/runtime.py

@@ -0,0 +1,74 @@
+"""Layer 3 — Runtime: attach coverage data to semantic nodes.
+
+Reads a ``coverage.py`` JSON report and maps executed/missing lines
+back to semantic nodes, populating ``RuntimeInfo``.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from sem.nodes import SemanticNode
+
+
+def _load_coverage_json(path: Path) -> dict:
+    """Load a coverage.py JSON report file."""
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def attach_coverage(
+    nodes: list[SemanticNode],
+    coverage_json: Path,
+    source_file: str,
+) -> list[SemanticNode]:
+    """Map coverage data onto semantic nodes.
+
+    Parameters
+    ----------
+    nodes
+        Semantic nodes for a single module (from ``parse_module``).
+    coverage_json
+        Path to the ``coverage.json`` report (``coverage json -o file``).
+    source_file
+        The source file key as it appears in the coverage report
+        (e.g. ``src/aco/express/pharmacy.py``).
+
+    Returns
+    -------
+    list[SemanticNode]
+        Same list, mutated in place.
+    """
+    report = _load_coverage_json(coverage_json)
+    files = report.get("files", {})
+    file_data = files.get(source_file)
+    if file_data is None:
+        return nodes
+
+    executed_lines: set[int] = set(file_data.get("executed_lines", []))
+    missing_lines: set[int] = set(file_data.get("missing_lines", []))
+
+    for node in nodes:
+        node_lines = set(range(node.span.start_line, node.span.end_line + 1))
+        hit_lines = node_lines & executed_lines
+        miss_lines = node_lines & missing_lines
+
+        if hit_lines:
+            node.runtime.hit = True
+            node.runtime.hit_count = len(hit_lines)
+
+        if miss_lines and not hit_lines:
+            node.runtime.hit = False
+
+    # Attach per-test contexts if available (coverage.py contexts)
+    contexts = file_data.get("contexts", {})
+    if contexts:
+        for line_str, test_list in contexts.items():
+            line = int(line_str)
+            for node in nodes:
+                if node.span.start_line <= line <= node.span.end_line:
+                    for t in test_list:
+                        if t and t not in node.runtime.tests_seen:
+                            node.runtime.tests_seen.append(t)
+
+    return nodes

src/sem/state.py

@@ -0,0 +1,95 @@
+"""Rolling state persistence for semantic coverage.
+
+Stores node-level tracking state (status, attempts, notes) in a JSON
+file that survives across runs.  Node IDs + source hashes make the
+state drift-resistant — edits that change a node's content invalidate
+its history rather than silently reusing stale data.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+from sem.nodes import SemanticNode
+
+
+def _default_entry() -> dict:
+    return {
+        "status": "unexplored",
+        "attempts": 0,
+        "last_result": None,
+        "source_hash": "",
+        "notes": [],
+    }
+
+
+def load_state(path: Path) -> dict[str, dict]:
+    """Load persisted state from a JSON file."""
+    if not path.exists():
+        return {}
+    return json.loads(path.read_text(encoding="utf-8"))
+
+
+def save_state(state: dict[str, dict], path: Path) -> None:
+    """Persist state to a JSON file."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(state, indent=2) + "\n", encoding="utf-8")
+
+
+def sync_state(nodes: list[SemanticNode], state: dict[str, dict]) -> dict[str, dict]:
+    """Merge current nodes into persisted state.
+
+    - New nodes get a fresh entry.
+    - Nodes whose ``source_hash`` changed are reset (code was edited).
+    - Existing nodes with matching hashes keep their history.
+    - Stale entries (nodes no longer in the AST) are removed.
+    """
+    updated: dict[str, dict] = {}
+
+    for node in nodes:
+        existing = state.get(node.node_id)
+        if existing is None:
+            updated[node.node_id] = _default_entry() | {
+                "source_hash": node.source_hash,
+            }
+        elif existing.get("source_hash") != node.source_hash:
+            # Source changed — reset but keep notes
+            updated[node.node_id] = _default_entry() | {
+                "source_hash": node.source_hash,
+                "notes": existing.get("notes", []),
+            }
+        else:
+            updated[node.node_id] = existing
+
+    return updated
+
+
+def apply_state(nodes: list[SemanticNode], state: dict[str, dict]) -> None:
+    """Push persisted state back onto node objects."""
+    for node in nodes:
+        entry = state.get(node.node_id)
+        if entry:
+            node.status = entry.get("status", "unexplored")
+            node.attempts = entry.get("attempts", 0)
+
+
+def record_attempt(
+    state: dict[str, dict],
+    node_id: str,
+    result: str,
+    note: str | None = None,
+) -> None:
+    """Record a test attempt against a node."""
+    entry = state.get(node_id)
+    if entry is None:
+        entry = _default_entry()
+        state[node_id] = entry
+    entry["attempts"] = entry.get("attempts", 0) + 1
+    entry["last_result"] = result
+    if result == "covered":
+        entry["status"] = "covered"
+    elif result == "partial":
+        entry["status"] = "partial"
+    if note:
+        entry.setdefault("notes", []).append(note)

tests/sem/test_enrich.py

@@ -0,0 +1,38 @@
+"""Tests for sem.enrich — Ruff and ty diagnostic attachment."""
+
+from __future__ import annotations
+
+from sem.enrich import _find_node
+from sem.nodes import NodeKind, SemanticNode, Span
+
+
+def _make_node(start: int, end: int, node_id: str = "n") -> SemanticNode:
+    return SemanticNode(
+        node_id=node_id,
+        kind=NodeKind.FUNCTION,
+        span=Span(start_line=start, end_line=end),
+        module="test",
+    )
+
+
+class TestFindNode:
+    def test_exact_match(self):
+        nodes = [_make_node(1, 5, "a"), _make_node(10, 20, "b")]
+        assert _find_node(nodes, 3) is nodes[0]
+        assert _find_node(nodes, 15) is nodes[1]
+
+    def test_tightest_span(self):
+        outer = _make_node(1, 20, "outer")
+        inner = _make_node(5, 10, "inner")
+        assert _find_node([outer, inner], 7) is inner
+
+    def test_no_match(self):
+        nodes = [_make_node(1, 5)]
+        assert _find_node(nodes, 100) is None
+
+    def test_boundary_lines(self):
+        node = _make_node(10, 20, "n")
+        assert _find_node([node], 10) is node
+        assert _find_node([node], 20) is node
+        assert _find_node([node], 9) is None
+        assert _find_node([node], 21) is None

tests/sem/test_parse.py

@@ -0,0 +1,170 @@
+"""Tests for sem.parse — AST to semantic nodes."""
+
+from __future__ import annotations
+
+import ast
+import textwrap
+
+from sem.nodes import NodeKind
+from sem.parse import parse_tree
+
+
+def _parse(source: str, module: str = "test_mod") -> list:
+    source = textwrap.dedent(source)
+    tree = ast.parse(source)
+    return parse_tree(tree, module, source)
+
+
+class TestFunctionNodes:
+    def test_top_level_function(self):
+        nodes = _parse("""
+            def greet(name):
+                return f"hi {name}"
+        """)
+        funcs = [n for n in nodes if n.kind == NodeKind.FUNCTION]
+        assert len(funcs) == 1
+        assert "greet" in funcs[0].node_id
+        assert funcs[0].module == "test_mod"
+
+    def test_method_inside_class(self):
+        nodes = _parse("""
+            class Foo:
+                def bar(self):
+                    pass
+        """)
+        classes = [n for n in nodes if n.kind == NodeKind.CLASS]
+        assert len(classes) == 1
+        # The method is nested inside the class context
+        methods = [n for n in nodes if n.kind in (NodeKind.FUNCTION, NodeKind.METHOD)]
+        assert len(methods) == 1
+        assert "Foo" in methods[0].node_id
+        assert "bar" in methods[0].node_id
+
+    def test_async_function(self):
+        nodes = _parse("""
+            async def fetch():
+                return 42
+        """)
+        funcs = [n for n in nodes if n.kind == NodeKind.FUNCTION]
+        assert len(funcs) == 1
+        assert "fetch" in funcs[0].node_id
+
+
+class TestBranchNodes:
+    def test_if_else(self):
+        nodes = _parse("""
+            def check(x):
+                if x > 0:
+                    return "pos"
+                else:
+                    return "neg"
+        """)
+        ifs = [n for n in nodes if n.kind == NodeKind.BRANCH_IF]
+        elses = [n for n in nodes if n.kind == NodeKind.BRANCH_ELSE]
+        assert len(ifs) == 1
+        assert len(elses) == 1
+
+    def test_elif(self):
+        nodes = _parse("""
+            def classify(x):
+                if x > 0:
+                    return "pos"
+                elif x == 0:
+                    return "zero"
+                else:
+                    return "neg"
+        """)
+        elifs = [n for n in nodes if n.kind == NodeKind.BRANCH_ELIF]
+        assert len(elifs) == 1
+
+    def test_match_case(self):
+        nodes = _parse("""
+            def route(cmd):
+                match cmd:
+                    case "start":
+                        pass
+                    case "stop":
+                        pass
+                    case _:
+                        pass
+        """)
+        cases = [n for n in nodes if n.kind == NodeKind.MATCH_CASE]
+        assert len(cases) == 3
+
+
+class TestExceptionNodes:
+    def test_except_handler(self):
+        nodes = _parse("""
+            def risky():
+                try:
+                    open("f")
+                except FileNotFoundError:
+                    pass
+                except ValueError:
+                    pass
+        """)
+        handlers = [n for n in nodes if n.kind == NodeKind.EXCEPT_HANDLER]
+        assert len(handlers) == 2
+
+
+class TestControlFlow:
+    def test_for_loop(self):
+        nodes = _parse("""
+            for i in range(10):
+                print(i)
+        """)
+        loops = [n for n in nodes if n.kind == NodeKind.FOR_LOOP]
+        assert len(loops) == 1
+
+    def test_while_loop(self):
+        nodes = _parse("""
+            while True:
+                break
+        """)
+        loops = [n for n in nodes if n.kind == NodeKind.WHILE_LOOP]
+        assert len(loops) == 1
+
+    def test_return_and_raise(self):
+        nodes = _parse("""
+            def f(x):
+                if x:
+                    return 1
+                raise ValueError("bad")
+        """)
+        returns = [n for n in nodes if n.kind == NodeKind.RETURN]
+        raises = [n for n in nodes if n.kind == NodeKind.RAISE]
+        assert len(returns) == 1
+        assert len(raises) == 1
+
+
+class TestNodeIdentity:
+    def test_node_id_is_stable(self):
+        """Same source should produce identical node IDs across parses."""
+        src = "def f():\n    return 1\n"
+        nodes_a = parse_tree(ast.parse(src), "mod", src)
+        nodes_b = parse_tree(ast.parse(src), "mod", src)
+        ids_a = [n.node_id for n in nodes_a]
+        ids_b = [n.node_id for n in nodes_b]
+        assert ids_a == ids_b
+
+    def test_source_hash_changes_with_content(self):
+        src_a = "def f():\n    return 1\n"
+        src_b = "def f():\n    return 2\n"
+        nodes_a = parse_tree(ast.parse(src_a), "mod", src_a)
+        nodes_b = parse_tree(ast.parse(src_b), "mod", src_b)
+        funcs_a = [n for n in nodes_a if n.kind == NodeKind.FUNCTION]
+        funcs_b = [n for n in nodes_b if n.kind == NodeKind.FUNCTION]
+        assert funcs_a[0].source_hash != funcs_b[0].source_hash
+
+    def test_node_id_no_line_numbers(self):
+        """Node IDs must not contain raw line numbers."""
+        nodes = _parse("""
+            def hello():
+                if True:
+                    return 1
+        """)
+        for node in nodes:
+            # Line numbers appear in span, not in node_id
+            assert node.span.start_line > 0
+            # node_id uses ordinal, not line number
+            assert ":L" not in node.node_id

tests/sem/test_plan.py

@@ -0,0 +1,76 @@
+"""Tests for sem.plan — prioritisation and target selection."""
+
+from __future__ import annotations
+
+from sem.nodes import NodeKind, RuntimeInfo, SemanticNode, Span
+from sem.plan import next_targets, rank_nodes, score_node
+
+
+def _node(
+    kind: NodeKind = NodeKind.BRANCH_IF,
+    hit: bool = False,
+    ruff: list[str] | None = None,
+    ty: list[str] | None = None,
+    attempts: int = 0,
+    module: str = "mod",
+    symbol_ctx: list[str] | None = None,
+) -> SemanticNode:
+    return SemanticNode(
+        node_id=f"{module}::{'::'.join(symbol_ctx or [])}::{kind.value}[0]",
+        kind=kind,
+        span=Span(start_line=1, end_line=5),
+        module=module,
+        symbol_context=symbol_ctx or [],
+        ruff_codes=ruff or [],
+        ty_codes=ty or [],
+        runtime=RuntimeInfo(hit=hit),
+        attempts=attempts,
+    )
+
+
+class TestScoring:
+    def test_uncovered_branch_scores_high(self):
+        n = _node(kind=NodeKind.BRANCH_IF, hit=False)
+        assert score_node(n, [n]) >= 5.0
+
+    def test_covered_node_penalised(self):
+        n = _node(hit=True)
+        assert score_node(n, [n]) < 0
+
+    def test_ty_codes_boost(self):
+        base = _node(kind=NodeKind.FUNCTION, hit=False)
+        with_ty = _node(kind=NodeKind.FUNCTION, hit=False, ty=["possibly-none"])
+        assert score_node(with_ty, [with_ty]) > score_node(base, [base])
+
+    def test_ruff_codes_boost(self):
+        base = _node(kind=NodeKind.FUNCTION, hit=False)
+        with_ruff = _node(kind=NodeKind.FUNCTION, hit=False, ruff=["B904"])
+        assert score_node(with_ruff, [with_ruff]) > score_node(base, [base])
+
+    def test_prior_failures_penalised(self):
+        fresh = _node(attempts=0)
+        stale = _node(attempts=3)
+        assert score_node(fresh, [fresh]) > score_node(stale, [stale])
+
+    def test_exception_handler_scores_high(self):
+        n = _node(kind=NodeKind.EXCEPT_HANDLER, hit=False)
+        assert score_node(n, [n]) >= 3.0
+
+
+class TestRanking:
+    def test_rank_order(self):
+        high = _node(kind=NodeKind.BRANCH_IF, hit=False, ty=["x"])
+        low = _node(kind=NodeKind.FUNCTION, hit=True)
+        ranked = rank_nodes([low, high])
+        assert ranked[0] is high
+
+    def test_next_targets_excludes_covered(self):
+        covered = _node(hit=True)
+        uncovered = _node(hit=False, kind=NodeKind.BRANCH_IF)
+        targets = next_targets([covered, uncovered], limit=5)
+        assert all(not t.runtime.hit for t in targets)
+
+    def test_next_targets_respects_limit(self):
+        nodes = [_node(hit=False, kind=NodeKind.BRANCH_IF) for _ in range(20)]
+        targets = next_targets(nodes, limit=3)
+        assert len(targets) <= 3

tests/sem/test_runtime.py

@@ -0,0 +1,78 @@
+"""Tests for sem.runtime — coverage data attachment."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from pathlib import Path
+
+from sem.nodes import NodeKind, SemanticNode, Span
+from sem.runtime import attach_coverage
+
+
+def _make_node(nid: str, start: int, end: int) -> SemanticNode:
+    return SemanticNode(
+        node_id=nid,
+        kind=NodeKind.BRANCH_IF,
+        span=Span(start_line=start, end_line=end),
+        module="pkg.mod",
+    )
+
+
+def _write_coverage(data: dict) -> Path:
+    p = Path(tempfile.mktemp(suffix=".json"))
+    p.write_text(json.dumps(data))
+    return p
+
+
+class TestAttachCoverage:
+    def test_marks_hit_nodes(self):
+        nodes = [_make_node("a", 1, 5), _make_node("b", 10, 15)]
+        cov = _write_coverage(
+            {
+                "files": {
+                    "src/pkg/mod.py": {
+                        "executed_lines": [1, 2, 3, 4, 5],
+                        "missing_lines": [10, 11, 12, 13, 14, 15],
+                    }
+                }
+            }
+        )
+        try:
+            attach_coverage(nodes, cov, "src/pkg/mod.py")
+            assert nodes[0].runtime.hit is True
+            assert nodes[0].runtime.hit_count == 5
+            assert nodes[1].runtime.hit is False
+        finally:
+            cov.unlink()
+
+    def test_missing_file_key_is_noop(self):
+        nodes = [_make_node("a", 1, 5)]
+        cov = _write_coverage({"files": {}})
+        try:
+            attach_coverage(nodes, cov, "nonexistent.py")
+            assert nodes[0].runtime.hit is False
+        finally:
+            cov.unlink()
+
+    def test_contexts_attached(self):
+        nodes = [_make_node("a", 1, 3)]
+        cov = _write_coverage(
+            {
+                "files": {
+                    "mod.py": {
+                        "executed_lines": [1, 2, 3],
+                        "missing_lines": [],
+                        "contexts": {
+                            "2": ["test_foo", "test_bar"],
+                        },
+                    }
+                }
+            }
+        )
+        try:
+            attach_coverage(nodes, cov, "mod.py")
+            assert "test_foo" in nodes[0].runtime.tests_seen
+            assert "test_bar" in nodes[0].runtime.tests_seen
+        finally:
+            cov.unlink()

tests/sem/test_state.py

@@ -0,0 +1,91 @@
+"""Tests for sem.state — rolling state persistence."""
+
+from __future__ import annotations
+
+from sem.nodes import NodeKind, SemanticNode, Span
+from sem.state import (
+    apply_state,
+    load_state,
+    record_attempt,
+    save_state,
+    sync_state,
+)
+
+
+def _node(nid: str, src_hash: str = "abc123") -> SemanticNode:
+    return SemanticNode(
+        node_id=nid,
+        kind=NodeKind.FUNCTION,
+        span=Span(start_line=1, end_line=5),
+        module="mod",
+        source_hash=src_hash,
+    )
+
+
+class TestPersistence:
+    def test_round_trip(self, tmp_path):
+        p = tmp_path / "state.json"
+        state = {"node_a": {"status": "covered", "attempts": 1}}
+        save_state(state, p)
+        loaded = load_state(p)
+        assert loaded["node_a"]["status"] == "covered"
+
+    def test_load_missing_file(self, tmp_path):
+        assert load_state(tmp_path / "nope.json") == {}
+
+
+class TestSync:
+    def test_new_node_gets_fresh_entry(self):
+        nodes = [_node("n1")]
+        result = sync_state(nodes, {})
+        assert "n1" in result
+        assert result["n1"]["status"] == "unexplored"
+
+    def test_unchanged_node_keeps_history(self):
+        nodes = [_node("n1", "abc")]
+        old = {"n1": {"status": "covered", "attempts": 3, "source_hash": "abc"}}
+        result = sync_state(nodes, old)
+        assert result["n1"]["status"] == "covered"
+        assert result["n1"]["attempts"] == 3
+
+    def test_changed_hash_resets_status(self):
+        nodes = [_node("n1", "new_hash")]
+        old = {"n1": {"status": "covered", "attempts": 3, "source_hash": "old_hash"}}
+        result = sync_state(nodes, old)
+        assert result["n1"]["status"] == "unexplored"
+        assert result["n1"]["attempts"] == 0
+
+    def test_stale_entries_removed(self):
+        nodes = [_node("n1")]
+        old = {"n1": {"source_hash": "abc123"}, "n_gone": {"source_hash": "x"}}
+        result = sync_state(nodes, old)
+        assert "n_gone" not in result
+
+
+class TestApplyState:
+    def test_pushes_state_to_nodes(self):
+        node = _node("n1")
+        state = {"n1": {"status": "partial", "attempts": 2}}
+        apply_state([node], state)
+        assert node.status == "partial"
+        assert node.attempts == 2
+
+
+class TestRecordAttempt:
+    def test_increments_attempts(self):
+        state = {"n1": {"status": "unexplored", "attempts": 0, "notes": []}}
+        record_attempt(state, "n1", "partial", "needs None input")
+        assert state["n1"]["attempts"] == 1
+        assert state["n1"]["last_result"] == "partial"
+        assert "needs None input" in state["n1"]["notes"]
+
+    def test_covered_sets_status(self):
+        state = {"n1": {"status": "unexplored", "attempts": 0, "notes": []}}
+        record_attempt(state, "n1", "covered")
+        assert state["n1"]["status"] == "covered"
+
+    def test_new_node_creates_entry(self):
+        state = {}
+        record_attempt(state, "new", "partial")
+        assert "new" in state
+        assert state["new"]["attempts"] == 1

uv.lock

@@ -3219,6 +3219,9 @@ rex = [
     { name = "pyarrow" },
     { name = "pydantic" },
 ]
+sem = [
+    { name = "coverage" },
+]
 
 [package.dev-dependencies]
 dev = [
@@ -3244,6 +3247,7 @@ requires-dist = [
     { name = "adlfs", marker = "extra == 'azure'", specifier = ">=2024.1.0" },
     { name = "azure-storage-blob", marker = "extra == 'azure'", specifier = ">=12.23.0" },
     { name = "boto3", marker = "extra == 'aws'", specifier = ">=1.35.0" },
+    { name = "coverage", marker = "extra == 'sem'", specifier = ">=7.13.4" },
     { name = "cryptography", marker = "extra == 'api'", specifier = ">=46.0.5" },
     { name = "databricks-bundles", marker = "extra == 'lake'", specifier = ">=0.295.0" },
     { name = "databricks-sdk", marker = "extra == 'lake'", specifier = ">=0.85.0" },
@@ -3314,7 +3318,7 @@ requires-dist = [
     { name = "uvicorn", marker = "extra == 'cli'", specifier = ">=0.41.0" },
     { name = "xlrd", specifier = ">=2.0.2" },
 ]
-provides-extras = ["conf", "aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "opps", "pfs", "rex", "perf", "lake", "aws", "gcp", "azure", "all"]
+provides-extras = ["conf", "aco", "api", "bcda", "bib", "bls", "ccw", "cli", "cms", "opps", "pfs", "rex", "perf", "sem", "lake", "aws", "gcp", "azure", "all"]
 
 [package.metadata.requires-dev]
 dev = [