Index

axm_echo

axm-echo.

Similarity & echo detection over code corpora (numpy/scikit-learn).

EchoCodeTool

Bases: AXMTool

Detect cross-package code echoes (intent-equivalent duplicates).

Registered as echo_code via the axm.tools entry point.

Source code in packages/axm-echo/src/axm_echo/tools.py

class EchoCodeTool(AXMTool):
    """Detect cross-package code echoes (intent-equivalent duplicates).

    Registered as ``echo_code`` via the ``axm.tools`` entry point.
    """

    agent_hint = (
        "Find intent-equivalent duplicate symbols across packages (clusters "
        "from cross-package docstring similarity, anti-signals applied)."
    )
    domain = "echo"
    tags = frozenset({"duplicate", "similarity", "echo", "clustering"})

    @property
    def name(self) -> str:
        """Return the tool name for registry lookup."""
        return "echo_code"

    def execute(
        self,
        *,
        backend: Backend = "st",
        threshold: float = PAIR_THRESHOLD,
        top_n: int = _DEFAULT_TOP_N,
        max_cluster_size: int = MAX_CLUSTER_SIZE,
        **kwargs: object,
    ) -> ToolResult:
        """Cluster cross-package echoes over the configured corpus.

        Args:
            backend: Embedding backend -- ``"st"`` (neural MiniLM, the
                in-process default) or ``"tfidf"`` (pure CPU, no torch).
            threshold: Minimum cosine for a candidate pair.
            top_n: Show at most this many of the nearest *non-acknowledged*
                clusters; the total count stays visible in the metadata.
            max_cluster_size: Reject components larger than this as union-find
                over-merges (structural conformity, not a duplicate echo).

        Returns:
            ToolResult with the bounded ``clusters`` (each carrying a
            ``cluster_hash``), ``parallel_api`` and ``boilerplate`` (demoted
            pairs), the live/shown/actionable counts, and ``stale_acknowledged``.
        """
        if backend not in ("st", "tfidf"):
            return ToolResult(
                success=False,
                error=f"Invalid backend {backend!r}; must be 'st' or 'tfidf'",
            )
        if not 0.0 <= threshold <= 1.0:
            return ToolResult(
                success=False,
                error=f"threshold must be a cosine in [0.0, 1.0], got {threshold}",
            )
        if top_n < 1:
            return ToolResult(success=False, error=f"top_n must be >= 1, got {top_n}")
        if max_cluster_size < 1:
            return ToolResult(
                success=False,
                error=f"max_cluster_size must be >= 1, got {max_cluster_size}",
            )

        try:
            return self._run(
                backend=backend,
                threshold=threshold,
                top_n=top_n,
                max_cluster_size=max_cluster_size,
            )
        except Exception as exc:  # noqa: BLE001 — final tool boundary
            logger.warning("EchoCodeTool failed: %s", exc, exc_info=True)
            return ToolResult(
                success=False,
                error=str(exc),
                hint=(
                    "Clustering failed over the monorepo corpus. Retry with "
                    "backend='tfidf' (no torch) if the neural backend is "
                    "unavailable, or check the corpus is reachable."
                ),
            )

    def _run(
        self,
        *,
        backend: Backend,
        threshold: float,
        top_n: int,
        max_cluster_size: int,
    ) -> ToolResult:
        """Execute the corpus -> embed -> pairs -> split -> cluster pipeline."""
        symbols = [
            s
            for s in extract_monorepo()
            if str(s.get("doc_full", "")).strip() and not is_trivial_accessor(s)
        ]
        if len(symbols) < _MIN_CORPUS:
            return ToolResult(
                success=True,
                data=self._empty_data(len(symbols)),
                text=self._render_text(
                    [],
                    [],
                    [],
                    corpus=len(symbols),
                    counts={"total": 0, "actionable": 0, "stale": 0},
                ),
            )

        texts = [str(s["embed_text"]) for s in symbols]
        packages = [str(s["package"]) for s in symbols]
        matrix = embed(texts, backend=backend)

        pairs = cross_pairs(matrix, packages, threshold=threshold)
        generic = generic_docs(symbols)
        dupes, parallel, boilerplate = split_pairs(pairs, symbols, generic)

        clusters = self._build_clusters(dupes, symbols, max_cluster_size)
        waivers, waiver_errors = self._load_waivers()
        mark_acknowledged(clusters, waivers)
        stale = stale_acknowledged(clusters, waivers)

        actionable = [c for c in clusters if not c.get("acknowledged")]
        shown = actionable[:top_n]
        parallel_entries = _pair_entries(parallel, symbols)
        boilerplate_entries = _pair_entries(boilerplate, symbols)

        data = {
            "corpus_size": len(symbols),
            "clusters": shown,
            "cluster_count": len(clusters),
            "actionable_count": len(actionable),
            "shown_count": len(shown),
            "parallel_api": parallel_entries,
            "boilerplate": boilerplate_entries,
            "stale_acknowledged": stale,
            "acknowledged_errors": waiver_errors,
        }
        text = self._render_text(
            shown,
            parallel_entries,
            boilerplate_entries,
            corpus=len(symbols),
            counts={
                "total": len(clusters),
                "actionable": len(actionable),
                "stale": len(stale),
            },
        )
        return ToolResult(success=True, data=data, text=text)

    @staticmethod
    def _load_waivers() -> tuple[list[dict[str, str]], list[str]]:
        """Read ``[[tool.axm-echo.acknowledged]]`` from the scan-root pyproject.

        The waiver lives in the pyproject of the *scan root* -- the first
        :func:`load_scope` root (a documented ownership choice for the
        cross-package case; cf. SPEC-similarity-echo §5bis). Each entry is
        validated; malformed entries are skipped and reported, never raised.

        Returns:
            ``(valid_waivers, errors)`` where each valid waiver is a
            ``{"hash", "reason"}`` dict and ``errors`` lists schema messages.
        """
        roots = load_scope()
        if not roots:
            return [], []
        section = _read_acknowledged_section(roots[0])
        return _partition_waivers(section)

    @staticmethod
    def _empty_data(corpus_size: int) -> dict[str, object]:
        """The data payload for a corpus too small to compare."""
        return {
            "corpus_size": corpus_size,
            "clusters": [],
            "cluster_count": 0,
            "actionable_count": 0,
            "shown_count": 0,
            "parallel_api": [],
            "boilerplate": [],
            "stale_acknowledged": [],
            "acknowledged_errors": [],
        }

    @staticmethod
    def _build_clusters(
        dupes: list[Pair], symbols: list[SymbolDict], max_cluster_size: int
    ) -> list[dict[str, object]]:
        """Connected components of the duplicate pairs, serialized + scored.

        Each serialized cluster carries a ``cluster_hash`` over its members'
        ``(package, qualname)`` so the waiver mechanism can address it.
        """
        components = cluster_pairs(dupes, max_cluster_size=max_cluster_size)
        clusters: list[dict[str, object]] = []
        for members in components:
            entry: dict[str, object] = {
                "size": len(members),
                "score": round(_max_pair_score(members, dupes), 4),
                "members": [_member(symbols[i]) for i in members],
            }
            entry["cluster_hash"] = cluster_hash(entry, key_fields=_ECHO_KEY_FIELDS)
            clusters.append(entry)
        clusters.sort(key=lambda c: cast("float", c["score"]), reverse=True)
        return clusters

    @staticmethod
    def _render_text(
        clusters: list[dict[str, object]],
        parallel: list[PairEntry],
        boilerplate: list[PairEntry],
        *,
        corpus: int,
        counts: Mapping[str, int],
    ) -> str:
        """Render the echo report as compact text for token-efficient MCP output.

        ``counts`` carries ``total`` (live clusters), ``actionable`` (live
        non-acknowledged) and ``stale`` (orphan waivers) for the header.
        """
        total = counts.get("total", 0)
        actionable = counts.get("actionable", 0)
        stale = counts.get("stale", 0)
        header = (
            f"echo_code | {total} clusters, {len(clusters)} shown "
            f"({actionable} actionable) | corpus {corpus} symbols | "
            f"{len(parallel)} parallel-API · {len(boilerplate)} boilerplate (demoted)"
        )
        if stale:
            header += f" | {stale} stale waiver(s)"
        if not clusters:
            return header
        lines = [header, ""]
        for idx, cluster in enumerate(clusters, start=1):
            lines.append(
                f"cluster {idx}  sim={cast('float', cluster['score']):.3f}  "
                f"({cluster['size']} symbols)"
            )
            members = cluster["members"]
            if not isinstance(members, list):
                continue
            for member in members:
                lines.append(
                    f"  {member['qualname']}  [{member['package']}]  "
                    f"“{member['doc_first_line']}”"
                )
        return "\n".join(lines)

name property

Return the tool name for registry lookup.

execute(*, backend='st', threshold=PAIR_THRESHOLD, top_n=_DEFAULT_TOP_N, max_cluster_size=MAX_CLUSTER_SIZE, **kwargs)

Cluster cross-package echoes over the configured corpus.

Parameters:

Name	Type	Description	Default
backend	Backend	Embedding backend -- "st" (neural MiniLM, the in-process default) or "tfidf" (pure CPU, no torch).	'st'
threshold	float	Minimum cosine for a candidate pair.	PAIR_THRESHOLD
top_n	int	Show at most this many of the nearest non-acknowledged clusters; the total count stays visible in the metadata.	_DEFAULT_TOP_N
max_cluster_size	int	Reject components larger than this as union-find over-merges (structural conformity, not a duplicate echo).	MAX_CLUSTER_SIZE

Returns:

Type	Description
ToolResult	ToolResult with the bounded clusters (each carrying a
ToolResult	cluster_hash), parallel_api and boilerplate (demoted
ToolResult	pairs), the live/shown/actionable counts, and stale_acknowledged.

Source code in packages/axm-echo/src/axm_echo/tools.py

def execute(
    self,
    *,
    backend: Backend = "st",
    threshold: float = PAIR_THRESHOLD,
    top_n: int = _DEFAULT_TOP_N,
    max_cluster_size: int = MAX_CLUSTER_SIZE,
    **kwargs: object,
) -> ToolResult:
    """Cluster cross-package echoes over the configured corpus.

    Args:
        backend: Embedding backend -- ``"st"`` (neural MiniLM, the
            in-process default) or ``"tfidf"`` (pure CPU, no torch).
        threshold: Minimum cosine for a candidate pair.
        top_n: Show at most this many of the nearest *non-acknowledged*
            clusters; the total count stays visible in the metadata.
        max_cluster_size: Reject components larger than this as union-find
            over-merges (structural conformity, not a duplicate echo).

    Returns:
        ToolResult with the bounded ``clusters`` (each carrying a
        ``cluster_hash``), ``parallel_api`` and ``boilerplate`` (demoted
        pairs), the live/shown/actionable counts, and ``stale_acknowledged``.
    """
    if backend not in ("st", "tfidf"):
        return ToolResult(
            success=False,
            error=f"Invalid backend {backend!r}; must be 'st' or 'tfidf'",
        )
    if not 0.0 <= threshold <= 1.0:
        return ToolResult(
            success=False,
            error=f"threshold must be a cosine in [0.0, 1.0], got {threshold}",
        )
    if top_n < 1:
        return ToolResult(success=False, error=f"top_n must be >= 1, got {top_n}")
    if max_cluster_size < 1:
        return ToolResult(
            success=False,
            error=f"max_cluster_size must be >= 1, got {max_cluster_size}",
        )

    try:
        return self._run(
            backend=backend,
            threshold=threshold,
            top_n=top_n,
            max_cluster_size=max_cluster_size,
        )
    except Exception as exc:  # noqa: BLE001 — final tool boundary
        logger.warning("EchoCodeTool failed: %s", exc, exc_info=True)
        return ToolResult(
            success=False,
            error=str(exc),
            hint=(
                "Clustering failed over the monorepo corpus. Retry with "
                "backend='tfidf' (no torch) if the neural backend is "
                "unavailable, or check the corpus is reachable."
            ),
        )

Symbol dataclass

A single public symbol projected into the corpus.

Source code in packages/axm-echo/src/axm_echo/corpus.py

@dataclass(frozen=True)
class Symbol:
    """A single public symbol projected into the corpus."""

    qualname: str
    name: str
    package: str
    workspace: str
    kind: str  # "function" | "class"
    signature: str
    doc_first_line: str
    doc_full: str
    body_norm: str
    path: str
    line: int

    @property
    def has_doc(self) -> bool:
        """Whether the symbol carries a non-empty docstring."""
        return bool(self.doc_full.strip())

    @property
    def embed_text(self) -> str:
        """Unified embed text: docstring if present, else code/signature."""
        if self.doc_full.strip():
            return f"{self.signature}\n{self.doc_full}".strip()
        return f"{self.signature}\n{self.body_norm}".strip()

    def as_dict(self) -> SymbolDict:
        """Flat dict view (qualname, package, signature, embed_text, ...)."""
        return {
            "qualname": self.qualname,
            "name": self.name,
            "package": self.package,
            "workspace": self.workspace,
            "kind": self.kind,
            "signature": self.signature,
            "doc_first_line": self.doc_first_line,
            "doc_full": self.doc_full,
            "body_norm": self.body_norm,
            "embed_text": self.embed_text,
            "has_doc": self.has_doc,
            "path": self.path,
            "line": self.line,
        }

embed_text property

Unified embed text: docstring if present, else code/signature.

has_doc property

Whether the symbol carries a non-empty docstring.

as_dict()

Flat dict view (qualname, package, signature, embed_text, ...).

Source code in packages/axm-echo/src/axm_echo/corpus.py

def as_dict(self) -> SymbolDict:
    """Flat dict view (qualname, package, signature, embed_text, ...)."""
    return {
        "qualname": self.qualname,
        "name": self.name,
        "package": self.package,
        "workspace": self.workspace,
        "kind": self.kind,
        "signature": self.signature,
        "doc_first_line": self.doc_first_line,
        "doc_full": self.doc_full,
        "body_norm": self.body_norm,
        "embed_text": self.embed_text,
        "has_doc": self.has_doc,
        "path": self.path,
        "line": self.line,
    }

code_tokens(src)

Tokenize code text for the TF-IDF backend.

Lowercases identifiers, splits camelCase and snake_case into sub-tokens, and appends structural keyword hints weighted by frequency so control-flow shape contributes to the vector.

Source code in packages/axm-echo/src/axm_echo/embedding.py

def code_tokens(src: str) -> list[str]:
    """Tokenize code text for the TF-IDF backend.

    Lowercases identifiers, splits ``camelCase`` and ``snake_case`` into
    sub-tokens, and appends structural keyword hints weighted by frequency
    so control-flow shape contributes to the vector.
    """
    tokens: list[str] = []
    for ident in _IDENT_RE.findall(src):
        tokens.append(ident.lower())
        parts = re.split(r"(?<=[a-z])(?=[A-Z])|_", ident)
        tokens.extend(p.lower() for p in parts if p)
    for kw in _STRUCTURAL_KEYWORDS:
        tokens.extend([kw] * src.count(kw))
    return tokens

config_path()

Return the path to the echo config (~/axm/echo.toml).

Does not check existence; callers degrade gracefully when absent.

Source code in packages/axm-echo/src/axm_echo/scope.py

def config_path() -> Path:
    """Return the path to the echo config (``~/axm/echo.toml``).

    Does not check existence; callers degrade gracefully when absent.
    """
    return Path.home() / _CONFIG_RELATIVE

discover_package_roots()

Discover every package directory across the configured scope.

Walks each workspace root from :func:axm_echo.scope.load_scope, covering both the <ws>/packages/<pkg> convention and the flat other/<pkg> layout. The set of roots is data-driven (no frozen package list), so newly added packages are picked up automatically (AC4).

Returns:

Type	Description
list[Path]	Sorted, de-duplicated package directories.

Source code in packages/axm-echo/src/axm_echo/corpus.py

def discover_package_roots() -> list[Path]:
    """Discover every package directory across the configured scope.

    Walks each workspace root from :func:`axm_echo.scope.load_scope`,
    covering both the ``<ws>/packages/<pkg>`` convention and the flat
    ``other/<pkg>`` layout. The set of roots is data-driven (no frozen
    package list), so newly added packages are picked up automatically
    (AC4).

    Returns:
        Sorted, de-duplicated package directories.
    """
    seen: set[Path] = set()
    roots: list[Path] = []
    for workspace_root in load_scope():
        for pkg in _packages_in_workspace(workspace_root):
            if pkg not in seen:
                seen.add(pkg)
                roots.append(pkg)
    return sorted(roots)

embed(texts, *, backend='tfidf')

Embed texts into a dense matrix via the chosen backend.

Parameters:

Name	Type	Description	Default
texts	Sequence[str]	Texts to embed (one row per text).	required
backend	Backend	"tfidf" (code, scikit-learn) or "st" (MiniLM, the default neural backend). The st backend lazily imports torch for first-call latency; tfidf never loads it.	'tfidf'

Returns:

Type	Description
NDArray[float64]	A (len(texts), dim) float matrix. Rows are not guaranteed
NDArray[float64]	normalized; neighbors normalizes internally.

Raises:

Type	Description
ValueError	If backend is not a registered backend.

Source code in packages/axm-echo/src/axm_echo/embedding.py

def embed(texts: Sequence[str], *, backend: Backend = "tfidf") -> NDArray[np.float64]:
    """Embed ``texts`` into a dense matrix via the chosen backend.

    Args:
        texts: Texts to embed (one row per text).
        backend: ``"tfidf"`` (code, scikit-learn) or ``"st"`` (MiniLM, the
            default neural backend). The ``st`` backend lazily imports torch
            for first-call latency; ``tfidf`` never loads it.

    Returns:
        A ``(len(texts), dim)`` float matrix. Rows are not guaranteed
        normalized; ``neighbors`` normalizes internally.

    Raises:
        ValueError: If ``backend`` is not a registered backend.
    """
    try:
        fn = _BACKENDS[backend]
    except KeyError:
        valid = ", ".join(sorted(_BACKENDS))
        msg = f"unknown backend {backend!r}; expected one of: {valid}"
        raise ValueError(msg) from None
    return fn(texts)

extract_monorepo()

Extract public symbols across every discovered package (corpus).

Returns:

Type	Description
list[SymbolDict]	The concatenated symbol corpus for the configured scope.

Source code in packages/axm-echo/src/axm_echo/corpus.py

def extract_monorepo() -> list[SymbolDict]:
    """Extract public symbols across every discovered package (corpus).

    Returns:
        The concatenated symbol corpus for the configured scope.
    """
    out: list[SymbolDict] = []
    for pkg in discover_package_roots():
        out.extend(extract_package(pkg))
    return out

extract_package(pkg_root)

Extract every public function/class of a package via axm-ast.

"Public" follows the axm-ast convention: a symbol exported in the module's __all__ when present, else any module-level name without a leading underscore. Test files and empty __init__ files are skipped; unparseable files are silently ignored.

Parameters:

Name	Type	Description	Default
pkg_root	Path	Package directory (containing src/<pkg>/ or a flat source tree).	required

Returns:

Type	Description
list[SymbolDict]	One Symbol per public function/class, each carrying
list[SymbolDict]	qualname, signature, doc_first_line, doc_full,
list[SymbolDict]	body_norm and the derived embed_text.

Source code in packages/axm-echo/src/axm_echo/corpus.py

def extract_package(pkg_root: Path) -> list[SymbolDict]:
    """Extract every public function/class of a package via axm-ast.

    "Public" follows the axm-ast convention: a symbol exported in the
    module's ``__all__`` when present, else any module-level name without
    a leading underscore. Test files and empty ``__init__`` files are
    skipped; unparseable files are silently ignored.

    Args:
        pkg_root: Package directory (containing ``src/<pkg>/`` or a flat
            source tree).

    Returns:
        One ``Symbol`` per public function/class, each carrying
        ``qualname``, ``signature``, ``doc_first_line``, ``doc_full``,
        ``body_norm`` and the derived ``embed_text``.
    """
    package, workspace = _package_meta(pkg_root)
    symbols: list[Symbol] = []
    for py in _iter_source_files(pkg_root):
        try:
            mod = extract_module_info(py)
        except (OSError, ValueError):
            continue
        mod_qual = _module_qualname(py, pkg_root)
        symbols.extend(_module_symbols(mod, mod_qual, py, package, workspace))
    return [s.as_dict() for s in symbols]

flatten_body(body)

Flatten compound statements (with/if/for/while/try) into their inner body.

Source code in packages/axm-echo/src/axm_echo/structural.py

def flatten_body(body: list[ast.stmt]) -> list[ast.stmt]:
    """Flatten compound statements (with/if/for/while/try) into their inner body."""
    out: list[ast.stmt] = []
    for stmt in body:
        match stmt:
            case ast.With() | ast.AsyncWith():
                out.extend(flatten_body(stmt.body))
            case ast.If() | ast.For() | ast.While() | ast.AsyncFor():
                out.extend(flatten_body(stmt.body))
                out.extend(flatten_body(stmt.orelse))
            case ast.Try():
                out.extend(flatten_body(stmt.body))
                for handler in stmt.handlers:
                    out.extend(flatten_body(handler.body))
                out.extend(flatten_body(stmt.orelse))
                out.extend(flatten_body(stmt.finalbody))
            case _:
                out.append(stmt)
    return out

jaccard_similarity(a, b)

Jaccard similarity between two sets (1.0 when both are empty).

Source code in packages/axm-echo/src/axm_echo/structural.py

def jaccard_similarity(
    a: set[str] | frozenset[str], b: set[str] | frozenset[str]
) -> float:
    """Jaccard similarity between two sets (1.0 when both are empty)."""
    if not a and not b:
        return 1.0
    if not a or not b:
        return 0.0
    inter = len(a & b)
    union = len(a | b)
    return inter / union if union else 0.0

load_scope()

Return the workspace roots to scan, with graceful degradation.

Reads workspace_roots from ~/axm/echo.toml when present and well-formed. On any failure (file absent, unreadable, invalid TOML, missing or empty workspace_roots) returns [Path.cwd()] so the caller still has the current workspace to scan -- never an exception.

Returns:

Type	Description
list[Path]	Resolved, de-duplicated workspace root paths. Always non-empty.

Source code in packages/axm-echo/src/axm_echo/scope.py

def load_scope() -> list[Path]:
    """Return the workspace roots to scan, with graceful degradation.

    Reads ``workspace_roots`` from ``~/axm/echo.toml`` when present and
    well-formed. On any failure (file absent, unreadable, invalid TOML,
    missing or empty ``workspace_roots``) returns ``[Path.cwd()]`` so the
    caller still has the current workspace to scan -- never an exception.

    Returns:
        Resolved, de-duplicated workspace root paths. Always non-empty.
    """
    roots = _read_configured_roots()
    if not roots:
        return [Path.cwd().resolve()]
    return roots

neighbors(query, matrix, *, k=10, threshold=None)

Cosine top-k nearest rows to query (exact brute-force matmul).

Parameters:

Name	Type	Description	Default
query	NDArray[float64]	A single query vector of shape (dim,).	required
matrix	NDArray[float64]	Corpus matrix of shape (n, dim).	required
k	int	Maximum number of neighbors to return.	10
threshold	float | None	If set, drop neighbors with cosine below this value.	None

Returns:

Type	Description
list[tuple[int, float]]	A list of (row_index, cosine_score) pairs sorted by score
list[tuple[int, float]]	descending, length <= k. The query is not excluded from the
list[tuple[int, float]]	corpus; if query is a row of matrix it appears at score 1.0.

Source code in packages/axm-echo/src/axm_echo/embedding.py

def neighbors(
    query: NDArray[np.float64],
    matrix: NDArray[np.float64],
    *,
    k: int = 10,
    threshold: float | None = None,
) -> list[tuple[int, float]]:
    """Cosine top-k nearest rows to ``query`` (exact brute-force matmul).

    Args:
        query: A single query vector of shape ``(dim,)``.
        matrix: Corpus matrix of shape ``(n, dim)``.
        k: Maximum number of neighbors to return.
        threshold: If set, drop neighbors with cosine below this value.

    Returns:
        A list of ``(row_index, cosine_score)`` pairs sorted by score
        descending, length ``<= k``. The query is not excluded from the
        corpus; if ``query`` is a row of ``matrix`` it appears at score 1.0.
    """
    norm_matrix = _l2_normalize(np.atleast_2d(matrix).astype(np.float64))
    norm_query = _l2_normalize(np.asarray(query, dtype=np.float64).reshape(1, -1))
    sims = (norm_query @ norm_matrix.T).ravel()

    order = np.argsort(-sims)
    results: list[tuple[int, float]] = []
    for idx in order:
        score = float(sims[idx])
        if threshold is not None and score < threshold:
            break
        results.append((int(idx), score))
        if len(results) >= k:
            break
    return results

normalize_dump(stmt)

Normalized single-statement dump (constants + name ids replaced).

Source code in packages/axm-echo/src/axm_echo/structural.py

def normalize_dump(stmt: ast.stmt) -> str | None:
    """Normalized single-statement dump (constants + name ids replaced)."""
    try:
        dump = ast.dump(stmt, annotate_fields=False)
    except Exception:  # noqa: BLE001
        return None
    dump = _CONSTANT_RE.sub("Constant(<C>)", dump)
    dump = _NAME_RE.sub("Name(<N>,", dump)
    return dump

statement_set(node)

Normalized stmt shapes (constants + name ids replaced) as a set.

Source code in packages/axm-echo/src/axm_echo/structural.py

def statement_set(node: ast.FunctionDef) -> frozenset[str]:
    """Normalized stmt shapes (constants + name ids replaced) as a set."""
    stmts: set[str] = set()
    for stmt in flatten_body(node.body):
        try:
            dump = ast.dump(stmt, annotate_fields=False)
        except Exception:  # noqa: BLE001, S112
            continue
        dump = _CONSTANT_RE.sub("Constant(<C>)", dump)
        dump = _NAME_RE.sub("Name(<N>,", dump)
        stmts.add(dump)
    return frozenset(stmts)