Skip to content

Flows

flows

FlowsTool — execution flow tracing with entry point detection.

FlowsTool

Bases: AXMTool

Trace execution flows and detect entry points in a Python package.

Registered as ast_flows via axm.tools entry point.

Source code in packages/axm-ast/src/axm_ast/tools/flows.py
Python
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
class FlowsTool(AXMTool):
    """Trace execution flows and detect entry points in a Python package.

    Registered as ``ast_flows`` via axm.tools entry point.
    """

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

    @safe_execute
    def execute(  # noqa: PLR0913
        self,
        *,
        path: str = ".",
        entry: str | None = None,
        max_depth: int = 5,
        cross_module: bool = False,
        detail: str = "trace",
        exclude_stdlib: bool = True,
        **kwargs: object,
    ) -> ToolResult:
        """Detect entry points or trace flows from a symbol.

        Without ``entry``: returns detected entry points.
        With ``entry``: traces BFS flow from that entry point.

        Args:
            path: Path to package directory.
            entry: Optional entry point name to trace from.
            max_depth: Maximum BFS depth for flow tracing.
            cross_module: Resolve imports and trace into external modules.
            detail: Level of detail — ``"trace"`` (default),
                ``"source"`` (includes function source code), or
                ``"compact"`` (tree-formatted string with
                box-drawing characters).
            exclude_stdlib: If False, include stdlib/builtin callees
                in the BFS trace.  Default True (exclude them).

        Returns:
            ToolResult with entry points or flow steps.  When tracing,
            ``data`` includes ``depth`` (actual max depth reached),
            ``count``, and ``truncated`` (True when frontier nodes at
            ``max_depth`` had unexpanded children).
            Returns ``success=False`` when the entry symbol is not found
            in the package.
        """
        pkg_path = Path(path).resolve()
        if not pkg_path.is_dir():
            return ToolResult(success=False, error=f"Not a directory: {pkg_path}")

        from axm_ast.core.cache import get_package
        from axm_ast.core.flows import (
            VALID_DETAILS,
            find_entry_points,
            format_flow_compact,
            trace_flow,
        )

        if detail not in VALID_DETAILS:
            return ToolResult(
                success=False,
                error=(
                    f"Invalid detail={detail!r}; must be one of {sorted(VALID_DETAILS)}"
                ),
            )

        pkg = get_package(pkg_path)

        if entry is not None:
            return self._trace_entry(
                pkg,
                entry,
                max_depth=max_depth,
                cross_module=cross_module,
                detail=detail,
                exclude_stdlib=exclude_stdlib,
                trace_flow=trace_flow,
                format_flow_compact=format_flow_compact,
            )

        return self._detect_entries(pkg, find_entry_points)

    def _trace_entry(  # noqa: PLR0913
        self,
        pkg: PackageInfo,
        entry: str,
        *,
        max_depth: int,
        cross_module: bool,
        detail: str,
        exclude_stdlib: bool,
        trace_flow: TraceFlow,
        format_flow_compact: FormatFlowCompact,
    ) -> ToolResult:
        """Trace a single entry point and build the result."""
        steps, truncated = trace_flow(
            pkg,
            entry,
            max_depth=max_depth,
            cross_module=cross_module,
            detail=detail,
            exclude_stdlib=exclude_stdlib,
        )
        actual_depth = max((s.depth for s in steps), default=0)

        if detail == "compact":
            compact = format_flow_compact(steps)
            compact_data: dict[str, object] = {
                "entry": entry,
                "compact": compact,
                "traces": compact,
                "depth": actual_depth,
                "cross_module": cross_module,
                "count": len(steps),
                "truncated": truncated,
            }
            return ToolResult(
                success=True,
                data=compact_data,
                text=render_compact_text(
                    entry=entry,
                    compact=compact,
                    depth=actual_depth,
                    cross_module=cross_module,
                    count=len(steps),
                    truncated=truncated,
                ),
            )

        step_dicts = _build_step_dicts(steps)
        data: dict[str, object] = {
            "entry": entry,
            "steps": step_dicts,
            "depth": actual_depth,
            "cross_module": cross_module,
            "count": len(steps),
            "truncated": truncated,
        }
        renderer = render_source_text if detail == "source" else render_trace_text
        return ToolResult(
            success=True,
            data=data,
            text=renderer(
                entry=entry,
                steps=step_dicts,
                depth=actual_depth,
                cross_module=cross_module,
                count=len(steps),
                truncated=truncated,
            ),
        )

    @staticmethod
    def _detect_entries(
        pkg: PackageInfo,
        find_entry_points: FindEntryPoints,
    ) -> ToolResult:
        """Detect entry points in the package."""
        entries = find_entry_points(pkg)
        entry_dicts: list[EntryPointDict] = [
            EntryPointDict(
                name=e.name,
                module=e.module,
                kind=e.kind,
                line=e.line,
                framework=e.framework,
            )
            for e in entries
        ]
        return ToolResult(
            success=True,
            data={
                "entry_points": entry_dicts,
                "count": len(entries),
            },
            text=render_entry_points_text(entry_dicts, count=len(entries)),
        )
name property

Return tool name for registry lookup.

execute(*, path='.', entry=None, max_depth=5, cross_module=False, detail='trace', exclude_stdlib=True, **kwargs)

Detect entry points or trace flows from a symbol.

Without entry: returns detected entry points. With entry: traces BFS flow from that entry point.

Parameters:

Name Type Description Default
path str

Path to package directory.

 '.'
entry str | None

Optional entry point name to trace from.

 None
max_depth int

Maximum BFS depth for flow tracing.

 5
cross_module bool

Resolve imports and trace into external modules.

 False
detail str

Level of detail — "trace" (default), "source" (includes function source code), or "compact" (tree-formatted string with box-drawing characters).

 'trace'
exclude_stdlib bool

If False, include stdlib/builtin callees in the BFS trace. Default True (exclude them).

 True

Returns:

Type Description
ToolResult

ToolResult with entry points or flow steps. When tracing,
ToolResult

data includes depth (actual max depth reached),
ToolResult

count, and truncated (True when frontier nodes at
ToolResult

max_depth had unexpanded children).
ToolResult

Returns success=False when the entry symbol is not found
ToolResult

in the package.
Source code in packages/axm-ast/src/axm_ast/tools/flows.py
Python
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
@safe_execute
def execute(  # noqa: PLR0913
    self,
    *,
    path: str = ".",
    entry: str | None = None,
    max_depth: int = 5,
    cross_module: bool = False,
    detail: str = "trace",
    exclude_stdlib: bool = True,
    **kwargs: object,
) -> ToolResult:
    """Detect entry points or trace flows from a symbol.

    Without ``entry``: returns detected entry points.
    With ``entry``: traces BFS flow from that entry point.

    Args:
        path: Path to package directory.
        entry: Optional entry point name to trace from.
        max_depth: Maximum BFS depth for flow tracing.
        cross_module: Resolve imports and trace into external modules.
        detail: Level of detail — ``"trace"`` (default),
            ``"source"`` (includes function source code), or
            ``"compact"`` (tree-formatted string with
            box-drawing characters).
        exclude_stdlib: If False, include stdlib/builtin callees
            in the BFS trace.  Default True (exclude them).

    Returns:
        ToolResult with entry points or flow steps.  When tracing,
        ``data`` includes ``depth`` (actual max depth reached),
        ``count``, and ``truncated`` (True when frontier nodes at
        ``max_depth`` had unexpanded children).
        Returns ``success=False`` when the entry symbol is not found
        in the package.
    """
    pkg_path = Path(path).resolve()
    if not pkg_path.is_dir():
        return ToolResult(success=False, error=f"Not a directory: {pkg_path}")

    from axm_ast.core.cache import get_package
    from axm_ast.core.flows import (
        VALID_DETAILS,
        find_entry_points,
        format_flow_compact,
        trace_flow,
    )

    if detail not in VALID_DETAILS:
        return ToolResult(
            success=False,
            error=(
                f"Invalid detail={detail!r}; must be one of {sorted(VALID_DETAILS)}"
            ),
        )

    pkg = get_package(pkg_path)

    if entry is not None:
        return self._trace_entry(
            pkg,
            entry,
            max_depth=max_depth,
            cross_module=cross_module,
            detail=detail,
            exclude_stdlib=exclude_stdlib,
            trace_flow=trace_flow,
            format_flow_compact=format_flow_compact,
        )

    return self._detect_entries(pkg, find_entry_points)

TraceFlow

Bases: Protocol

Signature of :func:axm_ast.core.flows.trace_flow.

Source code in packages/axm-ast/src/axm_ast/tools/flows.py
Python
27
28
29
30
31
32
33
34
35
36
37
38
39
class TraceFlow(Protocol):
    """Signature of :func:`axm_ast.core.flows.trace_flow`."""

    def __call__(  # noqa: PLR0913 — mirrors core.flows.trace_flow signature
        self,
        pkg: PackageInfo,
        entry: str,
        *,
        max_depth: int,
        cross_module: bool,
        detail: str,
        exclude_stdlib: bool,
    ) -> tuple[list[FlowStep], bool]: ...

render_compact_text(entry, compact, depth, cross_module, count, truncated)

Render compact text: header + raw format_flow_compact output.

Source code in packages/axm-ast/src/axm_ast/tools/flows_text.py
Python
141
142
143
144
145
146
147
148
149
150
151
152
153
def render_compact_text(  # noqa: PLR0913
    entry: str,
    compact: str,
    depth: int,
    cross_module: bool,
    count: int,
    truncated: bool,
) -> str:
    """Render compact text: header + raw format_flow_compact output."""
    params = _trace_params(entry, depth, cross_module, truncated)
    hdr = _header("compact", params, count, "steps")

    return f"{hdr}\n\n{compact}"

render_entry_points_text(entries, count)

Render entry points grouped by module with default elision.

Source code in packages/axm-ast/src/axm_ast/tools/flows_text.py
Python
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
def render_entry_points_text(entries: list[EntryPointDict], count: int) -> str:
    """Render entry points grouped by module with default elision."""
    hdr = _header("entry_points", "", count, "entries").replace(" |  | ", " | ")

    if not entries:
        return hdr

    by_module: dict[str, list[str]] = {}
    for e in entries:
        by_module.setdefault(e["module"], []).append(_format_entry(e))

    lines = [hdr, ""]
    for module, names in sorted(by_module.items()):
        lines.append(f"{module}: {' '.join(names)}")

    lines.append("")
    lines.append(
        "Legend: name (export, line=1) | name:LINE (export)"
        " | @FW name:LINE (decorator) | \u25b6name:LINE (main_guard)"
    )
    return "\n".join(lines)

render_source_text(entry, steps, depth, cross_module, count, truncated)

Render trace tree with inline source blocks.

Source code in packages/axm-ast/src/axm_ast/tools/flows_text.py
Python
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def render_source_text(  # noqa: PLR0913
    entry: str,
    steps: list[FlowStepDict],
    depth: int,
    cross_module: bool,
    count: int,
    truncated: bool,
) -> str:
    """Render trace tree with inline source blocks."""
    params = _trace_params(entry, depth, cross_module, truncated)
    hdr = _header("source", params, count, "steps")

    lines = [hdr, ""]
    for step in steps:
        lines.append(_format_step_line(step))
        source = step.get("source")
        if source:
            indent = "  " * (step["depth"] + 1)
            for src_line in source.splitlines():
                lines.append(f"{indent}{src_line}")

    return "\n".join(lines)

render_trace_text(entry, steps, depth, cross_module, count, truncated)

Render trace steps as indented tree text.

Source code in packages/axm-ast/src/axm_ast/tools/flows_text.py
Python
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
def render_trace_text(  # noqa: PLR0913
    entry: str,
    steps: list[FlowStepDict],
    depth: int,
    cross_module: bool,
    count: int,
    truncated: bool,
) -> str:
    """Render trace steps as indented tree text."""
    params = _trace_params(entry, depth, cross_module, truncated)
    hdr = _header("trace_flow", params, count, "steps")

    lines = [hdr, ""]
    for step in steps:
        lines.append(_format_step_line(step))

    return "\n".join(lines)