DEV Community

Dakota Wu
Dakota Wu

Posted on

Map Every Call Site Before You Split a God Function

The following walkthrough uses a synthetic order-processing module, not a production incident attributed to any employer. A coding agent proposed splitting process_order into retail, wholesale, and refund helpers after one overnight prompt. The diff looked tidy until a nightly CSV importer failed because it still passed a loosely typed dictionary. Mapping callers first would have shown that importer, an admin retry job, and a forgotten fax channel.

God functions hide disagreement among callers

A four-hundred-line callee is rarely one abstraction wearing an unfortunate name. Its callers usually disagree about required keys, default channels, and whether missing prices should raise or skip. An agent that reads only the function body will invent a narrower signature that matches the comments. The inventory below does not claim the module is well designed. It only records who still depends on the current shape.

You should treat a proposed split as a behavior change until every observed call site is listed. Characterization tests then pin those shapes so the first extract cannot quietly drop a branch. The extract itself should move one predicate, not three business lines, in a single commit.

Typical damage from an unmapped split looks like the list below.

  • A required channel argument appears, and the fax command still calls with one positional dict.
  • None for status becomes the string "open", which changes retry jobs that branch on identity.
  • Refunds move to a new module, while the importer still inspects the old tuple layout.
  • Dynamic getattr(orders, name) calls vanish from the agent's file-level rewrite.

A six-step workflow you can rerun

  1. Freeze the public function name and refuse renames until the caller map is green.
  2. Collect static call sites with an AST walk and a fallback ripgrep pass.
  3. Record argument shapes from fixtures, logs, or a short tracing wrapper.
  4. Turn each distinct shape into a characterization test that asserts today's result.
  5. Extract one branch whose predicate is already covered by those tests.
  6. Regenerate the caller map and fail the build if a site appears or disappears.

The loop is deliberately boring. Agents optimize for a clean file tree, while this loop optimizes for unchanged callers. Re-run the map after every agent session, not only after the extract you intended to accept.

Artifact: regenerate a caller map from AST

Save the following example script as tools/caller_map.py. It is labeled example code for a local inventory, not a measured production scanner.

"""Labeled example: inventory call sites for a function name."""
from __future__ import annotations

import ast
import json
import sys
from pathlib import Path

TARGET = "process_order"
SKIP_PARTS = {".venv", "node_modules", "__pycache__", "migrations"}


class CallCollector(ast.NodeVisitor):
    def __init__(self, path: str) -> None:
        self.path = path
        self.hits: list[dict] = []

    def visit_Call(self, node: ast.Call) -> None:
        name = None
        if isinstance(node.func, ast.Name) and node.func.id == TARGET:
            name = node.func.id
        elif isinstance(node.func, ast.Attribute) and node.func.attr == TARGET:
            name = node.func.attr
        if name:
            self.hits.append(
                {
                    "file": self.path,
                    "line": node.lineno,
                    "kwargs": [k.arg for k in node.keywords if k.arg],
                    "positional": len(node.args),
                    "starargs": any(isinstance(a, ast.Starred) for a in node.args),
                }
            )
        self.generic_visit(node)


def iter_python_files(root: Path):
    for path in root.rglob("*.py"):
        if any(part in SKIP_PARTS for part in path.parts):
            continue
        yield path


def main() -> int:
    root = Path(sys.argv[1] if len(sys.argv) > 1 else ".")
    rows: list[dict] = []
    for path in iter_python_files(root):
        try:
            tree = ast.parse(path.read_text(encoding="utf-8"))
        except SyntaxError as exc:
            rows.append({"file": str(path), "error": str(exc)})
            continue
        visitor = CallCollector(str(path))
        visitor.visit(tree)
        rows.extend(visitor.hits)
    Path("caller_map.json").write_text(json.dumps(rows, indent=2), encoding="utf-8")
    print(f"wrote {len(rows)} rows to caller_map.json")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())
Enter fullscreen mode Exit fullscreen mode

Run it beside a second, dumber net so dynamic calls are not silently dropped.

python tools/caller_map.py ./app
rg -n "process_order\s*\(" --glob "*.py" --glob "!tools/caller_map.py"
python -c "import json; print(len(json.load(open('caller_map.json'))))"
diff -u <(jq -S . caller_map.json) <(jq -S . caller_map.committed.json)
Enter fullscreen mode Exit fullscreen mode

The AST pass records keyword names and positional arity for each hit. The ripgrep pass catches comments, strings, and getattr patterns the tree walker will miss. Store both outputs in version control next to the characterization tests. A later agent diff cannot delete a caller without a failing check if that pair is committed.

What to keep in caller_map.json

Keep the committed map small enough to read in review, and reject extra fields that churn for no reason.

  • Keep file, line, kwargs, positional, and starargs.
  • Drop absolute timestamps, usernames, and generated tool versions.
  • Sort rows by file then line before you commit the file.
  • Treat a missing fax_retry.py row as a failed build, not a cleanup win.

Record argument shapes, not the agent's preferred types

Static hits are not enough, because two call sites can share a name and disagree about payload shape. Add a tracing wrapper in example tests or a staging importer, then dump distinct shapes to shapes.json.

# Labeled example: record observed kwargs without claiming production coverage.
from collections import Counter
import json
from functools import wraps

_shapes = Counter()


def trace_process_order(fn):
    @wraps(fn)
    def wrapper(*args, **kwargs):
        key = json.dumps(
            {
                "arity": len(args),
                "keys": sorted(kwargs),
                "channel": kwargs.get("channel"),
                "has_price": "price" in kwargs,
                "status": None
                if kwargs.get("status") is None
                else type(kwargs.get("status")).__name__,
            },
            sort_keys=True,
        )
        _shapes[key] += 1
        return fn(*args, **kwargs)

    return wrapper
Enter fullscreen mode Exit fullscreen mode

After a representative fixture run, persist the counter. A compact table for this synthetic module might look like the following, which you should replace with shapes you actually observed.

source arity keys channel status type count
csv_importer 1 order, channel retail str 412
admin_retry 1 order, channel, price wholesale NoneType 18
fax_command 1 order fax missing 2
refund_job 1 order, channel, refund retail str 77

The fax row is the one an agent will drop while renaming parameters to a closed set of channels. The inventory exists to make that row expensive to ignore. Counts in the table are synthetic labels for the example, not measurements from a live system.

Characterization tests built from the map

Each distinct shape becomes a test that calls today's process_order and asserts a frozen tuple of (status, total_cents, events). Do not assert internal helper names. Do not assert log wording unless a caller already parses those logs.

# Labeled example tests. Replace fixtures with recordings from your repo.
import json
from pathlib import Path

import pytest

from app.orders import process_order

SHAPES = json.loads(Path("shapes.json").read_text(encoding="utf-8"))


def _freeze(result):
    status, total_cents, events = result
    return status, total_cents, tuple(events)


@pytest.mark.parametrize("shape_id", sorted(SHAPES))
def test_observed_shape_still_returns_today(shape_id):
    payload = SHAPES[shape_id]["payload"]
    expected = tuple(SHAPES[shape_id]["frozen_result"])
    assert _freeze(process_order(**payload)) == expected


def test_fax_channel_is_still_a_caller():
    result = process_order(order={"id": "fax-1", "lines": []}, channel="fax")
    assert result[0] in {"skipped", "queued"}


def test_caller_map_has_not_shrunk():
    recorded = json.loads(Path("caller_map.json").read_text(encoding="utf-8"))
    files = {row["file"] for row in recorded if "line" in row}
    assert "app/importers/csv_orders.py" in files
    assert "app/management/fax_retry.py" in files
Enter fullscreen mode Exit fullscreen mode

If you lack a snapshot helper, write the frozen tuple next to the payload in shapes.json on the first green run, then commit the file. Later runs must not rewrite that file during an agent session. A one-line CI check that shapes.json is unchanged is cheaper than arguing about intent in review.

git diff --exit-code -- shapes.json caller_map.json
pytest -q tests/test_process_order_shapes.py
Enter fullscreen mode Exit fullscreen mode

Extract one branch after the map is green

Only after the tests pass should you move a single predicate. In this synthetic module the refund path is the safest first extract, because the caller map shows one job and the shape table shows one extra key.

# Before: one function owns every channel and the refund mutation.
def process_order(order, channel="retail", price=None, refund=None, status="open"):
    if refund:
        return _refund_inline(order, refund, status)
    if channel == "fax":
        return "queued", 0, ("fax-deferred",)
    # hundreds of retail and wholesale lines follow
    raise NotImplementedError("example remainder")


# After: same signature, refund body lives next door.
def process_order(order, channel="retail", price=None, refund=None, status="open"):
    if refund:
        return process_refund(order, refund=refund, status=status)
    if channel == "fax":
        return "queued", 0, ("fax-deferred",)
    raise NotImplementedError("example remainder")
Enter fullscreen mode Exit fullscreen mode

Keep the original function name, default values, and return tuple. The agent can rename internals only after this commit stays green for a full importer run. If the agent also wants to delete channel="fax", that is a product change and needs its own test and rollout, not a cleanup label.

Review checklist for the extract commit

  • The public signature is byte-for-byte unchanged, including defaults.
  • caller_map.json still lists every previously committed file.
  • shapes.json frozen tuples are unchanged for retail, wholesale, fax, and refund.
  • No rename of process_order landed in the same commit as the extract.
  • Ripgrep still finds the fax command, even if AST missed a dynamic helper.

Decision table for the next commit

Observation Allowed next commit Reject the agent diff
New helper, same signature, same frozen tuples Extract one covered branch Extra signature changes in the same PR
Caller map row count unchanged Proceed Silent deletion of fax_command
Shape keys unchanged Proceed price became required
AST miss but ripgrep hit on getattr Add a test, then extract Assume dynamic calls are dead
Tests pass, importer golden file drifts Stop and inspect events Explain it as formatting

Use the table in review comments instead of debating whether the agent meant well. The map either still describes production callers, or the cleanup is not a refactor.

Running the inventory off your laptop

The script and the tests above are ordinary Python, and they do not require any vendor. When the messy tree is too large for a local checkout, a disposable machine can still regenerate the map. A model may propose the one-branch extract against that copy after the characterization suite is green.

Disclosure: This article was prepared as part of MonkeyCode's product outreach. MonkeyCode currently offers free model access and a free server option that can run this inventory. Those options can host the caller-map script on a throwaway copy of the repository tree. Do not place production secrets, customer dumps, or live credentials on a shared free host. Treat model output as a draft extract that still has to pass the committed map.

If you already have CI, run caller_map.py in that pipeline and skip any hosted environment. The workflow remains useful if that hosted option is removed from the process entirely.

Limitations you should budget for

  • AST inventory misses getattr(module, name), exec, and calls assembled from strings.
  • Ripgrep over-counts comments, tests, and documentation that mention the function.
  • JSON shape keys collapse distinct values that happen to share a type name.
  • Tracing wrappers change timing and can hide race-only branches in messy modules.
  • Extracting one branch can still alter exception order if helpers start catching earlier.
  • Frozen tuples will not catch extra outbound HTTP unless you also stub those clients.
  • A free shared server is the wrong place for proprietary order data or signing keys.

None of these limitations are reasons to skip the map. They are reasons to keep the first extract small and to read the ripgrep misses by hand. If a call site cannot be expressed as JSON, write a dedicated test rather than widening the tracer until it lies.

Who should not use this approach

Skip this workflow if the module already has an honest public contract and a complete caller list in your type checker. Skip it if your goal is a ground-up rewrite rather than a behavior-preserving cleanup. Skip it if the only available compute is a shared free host and the repository contains secrets, because the inventory is not worth a credential leak. Skip it if you cannot freeze shapes.json against mutation during an agent session.

Teams that should use it are the ones receiving confident cleanup pull requests against functions that still serve importers, cron jobs, and one forgotten channel. The caller map will not make the god function elegant. It will make the first extract boring enough to merge.

Top comments (0)