Tools

tools

echo AXMTools -- cross-package echo detection and intent retrieval.

Two read-only AXMTools in the spirit of ast_dead_code (see axm_ast/tools/dead_code.py), sharing the corpus -> embed pipeline:

• :class:EchoCodeTool (echo_code) walks the configured monorepo scope, embeds every public documented symbol, finds cross-package pairs whose promises (docstrings) are semantically close, applies the v7 anti-signals, and returns the surviving duplicate clusters plus the demoted parallel-API / boilerplate buckets.
• :class:EchoCheckTool (echo_check) embeds a free-form intention and retrieves the top-k nearest public symbols across the whole monorepo, each tagged with a location verdict (reuse canonical / reuse in place / promotable). It does the retrieval, never the use/extend/nothing decision -- that is left to the calling agent.

Both are registered under the axm.tools entry point, so each is reachable as an MCP tool, an axm <name> CLI command, and a DAG tool_node for free (one declaration, three surfaces).

CandidateEntry

Bases: TypedDict

A retrieved symbol matching an intention, with its location verdict.

The verdict is purely a location tag (AC3) -- it never encodes a use/extend/nothing decision (AC4): a high score does not mean "use this", only "this is the closest existing promise". promotable flags a non-ingot candidate documented well enough to be worth canonicalising.

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

class CandidateEntry(TypedDict):
    """A retrieved symbol matching an intention, with its location verdict.

    The ``verdict`` is purely a *location* tag (AC3) -- it never encodes a
    use/extend/nothing decision (AC4): a high score does not mean "use this",
    only "this is the closest existing promise". ``promotable`` flags a
    non-ingot candidate documented well enough to be worth canonicalising.
    """

    score: float
    verdict: str
    promotable: bool
    qualname: str
    name: str
    package: str
    doc_first_line: str
    doc_full: str
    path: str
    line: int

ClusterEntry

Bases: TypedDict

A cross-package echo cluster (connected component of duplicate pairs).

cluster_hash is the waiver address over the members' (package, qualname); acknowledged is stamped when a live cluster is waived. Both are added during the run, hence total=False.

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

class ClusterEntry(TypedDict, total=False):
    """A cross-package echo cluster (connected component of duplicate pairs).

    ``cluster_hash`` is the waiver address over the members' ``(package,
    qualname)``; ``acknowledged`` is stamped when a live cluster is waived.
    Both are added during the run, hence ``total=False``.
    """

    size: int
    score: float
    members: list[MemberEntry]
    cluster_hash: str
    acknowledged: bool

EchoCheckTool

Bases: AXMTool

Retrieve the public symbols closest to a free-form intention.

Registered as echo_check via the axm.tools entry point. It embeds the intention, retrieves the top-k nearest documented symbols across the whole monorepo corpus (AC2), and tags each with a location verdict (AC3). Retrieval is decoupled from the use/extend/nothing decision (AC4): the tool returns ranked candidates + docstrings and leaves the call to the agent.

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

class EchoCheckTool(AXMTool):
    """Retrieve the public symbols closest to a free-form *intention*.

    Registered as ``echo_check`` via the ``axm.tools`` entry point. It embeds
    the intention, retrieves the top-k nearest documented symbols across the
    whole monorepo corpus (AC2), and tags each with a location verdict (AC3).
    Retrieval is decoupled from the use/extend/nothing decision (AC4): the tool
    returns ranked candidates + docstrings and leaves the call to the agent.
    """

    agent_hint = (
        "Before writing a helper, retrieve the closest existing symbols across "
        "the monorepo for an intention (top-k + docstrings + reuse/promote "
        "verdict). Decide use/extend/nothing yourself -- this only ranks."
    )
    domain = "echo"
    tags = frozenset({"reuse", "similarity", "echo", "retrieval"})

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

    def execute(
        self,
        *,
        intention: str = "",
        backend: Backend = "st",
        k: int = _CHECK_TOP_K,
        threshold: float = _CHECK_THRESHOLD,
        **kwargs: object,
    ) -> ToolResult:
        """Retrieve the top-k symbols closest to *intention* over the corpus.

        Args:
            intention: Free-form description of the behaviour to implement.
            backend: Embedding backend -- ``"st"`` (neural MiniLM, the
                in-process default) or ``"tfidf"`` (pure CPU, no torch).
            k: Maximum number of candidates to return.
            threshold: Minimum cosine for a candidate to be retrieved. Below it
                the candidate is dropped, so a novel intention returns an empty
                list rather than a spurious match (AC4).

        Returns:
            ToolResult with ``intention``, ``corpus_size`` and ``candidates``
            (ranked top-k, each carrying its docstrings and a location verdict).
        """
        try:
            return self._run(
                intention=intention, backend=backend, k=k, threshold=threshold
            )
        except Exception as exc:  # noqa: BLE001 — final tool boundary
            logger.warning("EchoCheckTool failed: %s", exc, exc_info=True)
            return ToolResult(success=False, error=str(exc))

    def _run(
        self, *, intention: str, backend: Backend, k: int, threshold: float
    ) -> ToolResult:
        """Execute the corpus -> embed -> retrieve -> verdict pipeline."""
        if not intention.strip():
            raise ValueError("intention must be a non-empty string")

        symbols = [
            s
            for s in extract_monorepo()
            if str(s.get("doc_full", "")).strip() and not is_trivial_accessor(s)
        ]
        if not symbols:
            return ToolResult(
                success=True,
                data={"intention": intention, "corpus_size": 0, "candidates": []},
                text=self._render_text(intention, [], corpus=0),
            )

        texts = [str(s["embed_text"]) for s in symbols]
        matrix = embed([intention, *texts], backend=backend)
        hits = neighbors(matrix[0], matrix[1:], k=k, threshold=threshold)

        candidates = [_candidate(symbols[idx], score) for idx, score in hits]
        data = {
            "intention": intention,
            "corpus_size": len(symbols),
            "candidates": candidates,
        }
        text = self._render_text(intention, candidates, corpus=len(symbols))
        return ToolResult(success=True, data=data, text=text)

    @staticmethod
    def _render_text(
        intention: str, candidates: list[CandidateEntry], *, corpus: int
    ) -> str:
        """Render the retrieval report as compact, token-efficient text."""
        header = (
            f"echo_check | “{intention}” | {len(candidates)} candidates | "
            f"corpus {corpus} symbols"
        )
        if not candidates:
            return f"{header}\n(no candidate above threshold — likely novel)"
        lines = [header, ""]
        for rank, cand in enumerate(candidates, start=1):
            promote = " (promotable→ingot)" if cand["promotable"] else ""
            lines.append(
                f"{rank}. {cand['qualname']}  [{cand['package']}]  "
                f"sim={cand['score']:.3f}  {cand['verdict']}{promote}"
            )
            lines.append(f'   "{cand["doc_first_line"]}"')
        return "\n".join(lines)

name property

Return the tool name for registry lookup.

execute(*, intention='', backend='st', k=_CHECK_TOP_K, threshold=_CHECK_THRESHOLD, **kwargs)

Retrieve the top-k symbols closest to intention over the corpus.

Parameters:

Name	Type	Description	Default
intention	str	Free-form description of the behaviour to implement.	''
backend	Backend	Embedding backend -- "st" (neural MiniLM, the in-process default) or "tfidf" (pure CPU, no torch).	'st'
k	int	Maximum number of candidates to return.	_CHECK_TOP_K
threshold	float	Minimum cosine for a candidate to be retrieved. Below it the candidate is dropped, so a novel intention returns an empty list rather than a spurious match (AC4).	_CHECK_THRESHOLD

Returns:

Type	Description
ToolResult	ToolResult with intention, corpus_size and candidates
ToolResult	(ranked top-k, each carrying its docstrings and a location verdict).

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

def execute(
    self,
    *,
    intention: str = "",
    backend: Backend = "st",
    k: int = _CHECK_TOP_K,
    threshold: float = _CHECK_THRESHOLD,
    **kwargs: object,
) -> ToolResult:
    """Retrieve the top-k symbols closest to *intention* over the corpus.

    Args:
        intention: Free-form description of the behaviour to implement.
        backend: Embedding backend -- ``"st"`` (neural MiniLM, the
            in-process default) or ``"tfidf"`` (pure CPU, no torch).
        k: Maximum number of candidates to return.
        threshold: Minimum cosine for a candidate to be retrieved. Below it
            the candidate is dropped, so a novel intention returns an empty
            list rather than a spurious match (AC4).

    Returns:
        ToolResult with ``intention``, ``corpus_size`` and ``candidates``
        (ranked top-k, each carrying its docstrings and a location verdict).
    """
    try:
        return self._run(
            intention=intention, backend=backend, k=k, threshold=threshold
        )
    except Exception as exc:  # noqa: BLE001 — final tool boundary
        logger.warning("EchoCheckTool failed: %s", exc, exc_info=True)
        return ToolResult(success=False, error=str(exc))

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."
            ),
        )

MemberEntry

Bases: TypedDict

A serialized symbol inside a cluster or a demoted pair.

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

class MemberEntry(TypedDict):
    """A serialized symbol inside a cluster or a demoted pair."""

    qualname: str
    name: str
    package: str
    doc_first_line: str
    path: str
    line: int

PairEntry

Bases: TypedDict

A demoted cross-package pair (parallel-API or boilerplate).

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

class PairEntry(TypedDict):
    """A demoted cross-package pair (parallel-API or boilerplate)."""

    score: float
    a: MemberEntry
    b: MemberEntry