DEV Community

Dakota Huang
Dakota Huang

Posted on

Canonicalize Mixed Returns Before You Rewrite One Function

Do not rewrite a messy function on the first pass.
Capture every input and output pair before you edit.
A two-column ledger is the cheapest oracle available today.

The problem this workflow actually solves

Silent contracts break even when the new diff looks tidy.
Callers depend on return shapes you never documented.
Exception types and message prefixes become accidental APIs.

A green unit suite often tests the happy path only.
Refactors fail on None, empty lists, and odd encodings.
Downstream code then fails while the function still works.

Mixed returns make this worse than a missing assertion.
One call returns a string. The next returns a tuple.
JSON, logs, and tests will not agree without a codec.

What an I/O ledger is

An I/O ledger is a JSONL file of recorded calls.
Each line stores arguments, keywords, and the returned value.
Failures store the exception type and a stable message stem.

The ledger is not a design document or type spec.
It is a characterization table built from real call sites.
You freeze that table, then change one function body.

JSON is the storage format. JSON is also the trap.
JSON has no tuple type and no raw bytes type.
Canonicalize those values before you trust any row.

When this method is the right tool

Use it when one function is tangled and heavily called.
Use it when tests exist but mock the wrong layer.
Skip it when the function is already a pure one-liner.

Do not start with a full-module rewrite or rename.
Do not extract helpers before the ledger stays green.
The smallest safe change remains a body-only rewrite.

Workflow

1. Pick one function, not a package

Choose a single function with mixed return types.
Prefer functions that parse, normalize, or classify input.
Leave class hierarchies and import graphs completely untouched.

Write the module path and callable name in notes.
Do not open a refactor branch at this stage.

2. Build a recorder, not a redesign

Wrap the function with a thin recording decorator.
Write each call as one JSONL object on disk.
Keep the original behavior completely unchanged during capture.

Canonicalize arguments and results before each disk write.
That step is the oracle. The decorator is only plumbing.

3. Drive the function from real entry points

Run the existing test suite against the wrapped function.
Run one fixture script if the current tests are thin.
Do not invent extra scenarios until the dump is complete.

Collect the paths your callers already exercise today.
Missing rows mean missing characterization, not extra features.
Stop collecting when repeated inputs stop adding new rows.

4. Turn the ledger into failing-closed tests

Generate pytest cases directly from the JSONL file.
Each row asserts return equality or exception class.
Compare message stems, not the full traceback text.

Decode tagged bytes back into bytes before the call.
Leave tuples as lists on the result side of JSON.
The production function may still return real tuples.

5. Freeze the ledger, then change one body

Commit the JSONL file and the generated tests together.
Rewrite only the function body and keep the signature frozen.
Run the ledger tests and diff nothing else in the patch.

If a row fails, the rewrite is not equivalent.
Restore the body, then split the change and repeat.
Do not edit the ledger to match a new guess.

Canonicalization rules

Apply these four rules before any row hits disk.
Skip one rule and the oracle will lie later.

  1. Convert tuples to lists before dump and compare.
  2. Encode bytes arguments as base64 with a type tag.
  3. Leave None, bools, ints, and strings unchanged.
  4. Keep exception class plus a 48-character message stem.

Do not serialize open files, sockets, or class instances.
Those values do not belong in a characterization ledger.
Pick a different seam if the function returns live handles.

Artifact: messy target, codec, recorder, tests

The listing below is a proposal, not a production run.
It uses the Python 3 standard library only.
Copy the files into a scratch directory and execute them.

# messy.py
"""Brownfield helper with mixed returns and mixed exceptions."""


def normalize_ref(raw, strict=False):
    if raw is None:
        if strict:
            raise TypeError("ref required")
        return None
    if isinstance(raw, bytes):
        raw = raw.decode("utf-8", errors="replace")
    text = str(raw).strip()
    if not text:
        if strict:
            raise ValueError("ref empty")
        return ""
    if text.startswith("id:"):
        body = text[3:].strip()
        if not body.isdigit():
            if strict:
                raise ValueError("ref id not numeric")
            return ("id", None)
        return ("id", int(body))
    if text.lower() in {"n/a", "na", "none", "null"}:
        return None
    if "," in text:
        parts = [p.strip() for p in text.split(",") if p.strip()]
        return parts or ""
    return text.lower()
Enter fullscreen mode Exit fullscreen mode
# codec.py
from __future__ import annotations

import base64
from typing import Any

BYTES_TAG = "__bytes_b64__"


def encode_value(value: Any) -> Any:
    if isinstance(value, bytes):
        return {BYTES_TAG: base64.b64encode(value).decode("ascii")}
    if isinstance(value, tuple):
        return [encode_value(v) for v in value]
    if isinstance(value, list):
        return [encode_value(v) for v in value]
    if isinstance(value, dict):
        return {str(k): encode_value(v) for k, v in value.items()}
    return value


def decode_value(value: Any) -> Any:
    if isinstance(value, dict) and set(value.keys()) == {BYTES_TAG}:
        return base64.b64decode(value[BYTES_TAG].encode("ascii"))
    if isinstance(value, list):
        return [decode_value(v) for v in value]
    if isinstance(value, dict):
        return {k: decode_value(v) for k, v in value.items()}
    return value


def encode_args(args: tuple[Any, ...], kwargs: dict[str, Any]):
    return [encode_value(a) for a in args], {
        k: encode_value(v) for k, v in kwargs.items()
    }


def decode_args(args: list[Any], kwargs: dict[str, Any]):
    return tuple(decode_value(a) for a in args), {
        k: decode_value(v) for k, v in kwargs.items()
    }


def message_stem(msg: str, n: int = 48) -> str:
    return " ".join(str(msg).split())[:n]
Enter fullscreen mode Exit fullscreen mode
# record_io.py
from __future__ import annotations

import json
import functools
from pathlib import Path
from typing import Any, Callable

from codec import encode_args, encode_value, message_stem

LEDGER = Path("io_ledger.jsonl")


def _dump(row: dict[str, Any]) -> None:
    with LEDGER.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(row, sort_keys=True) + "\n")


def record_io(fn: Callable) -> Callable:
    @functools.wraps(fn)
    def wrapper(*args: Any, **kwargs: Any) -> Any:
        enc_args, enc_kwargs = encode_args(args, kwargs)
        row: dict[str, Any] = {
            "fn": f"{fn.__module__}.{fn.__name__}",
            "args": enc_args,
            "kwargs": enc_kwargs,
        }
        try:
            result = fn(*args, **kwargs)
        except Exception as exc:
            row["ok"] = False
            row["exc_type"] = type(exc).__name__
            row["exc_stem"] = message_stem(exc)
            _dump(row)
            raise
        row["ok"] = True
        row["result"] = encode_value(result)
        _dump(row)
        return result

    return wrapper
Enter fullscreen mode Exit fullscreen mode
# seed_calls.py
from record_io import record_io
import messy

messy.normalize_ref = record_io(messy.normalize_ref)

SAMPLES = [
    (None, False),
    (None, True),
    ("", False),
    ("", True),
    ("  ", False),
    ("id:12", False),
    ("id:xx", False),
    ("id:xx", True),
    ("N/A", False),
    ("Foo", False),
    ("a, b, ", False),
    (b"Bar", False),
    (123, False),
]


def main() -> None:
    for raw, strict in SAMPLES:
        try:
            messy.normalize_ref(raw, strict=strict)
        except Exception:
            pass


if __name__ == "__main__":
    main()
Enter fullscreen mode Exit fullscreen mode
# test_io_ledger.py
import json
from pathlib import Path

import pytest

import messy
from codec import decode_args, encode_value, message_stem

LEDGER = Path("io_ledger.jsonl")


def load_rows():
    rows = []
    for line in LEDGER.read_text(encoding="utf-8").splitlines():
        if line.strip():
            rows.append(json.loads(line))
    return rows


@pytest.mark.parametrize("row", load_rows())
def test_ledger_row(row):
    args, kwargs = decode_args(row["args"], row["kwargs"])
    if row["ok"]:
        got = encode_value(messy.normalize_ref(*args, **kwargs))
        assert got == row["result"]
        return
    with pytest.raises(Exception) as caught:
        messy.normalize_ref(*args, **kwargs)
    assert type(caught.value).__name__ == row["exc_type"]
    assert message_stem(caught.value) == row["exc_stem"]
Enter fullscreen mode Exit fullscreen mode

Run the capture step before any body rewrite exists.
Then run the ledger tests as a closed characterization gate.

python seed_calls.py
python -m pytest test_io_ledger.py -q
Enter fullscreen mode Exit fullscreen mode

Expect one JSONL line per sample, including exception rows.
Open io_ledger.jsonl and confirm tuples stored as lists.
Confirm the b"Bar" sample stored a base64 type tag.

How to read a failing row

A failing row is a contract mismatch, not a test bug.
Read it in this order and change only one thing.

  1. Check ok, exc_type, and exc_stem before values.
  2. Diff result only after exception fields already match.
  3. Restore the body if the signature was not supposed to change.

Do not widen types to make a red row turn green.
Do not lowercase a new exception message to match the stem.
The ledger records the old contract, not your preferred one.

Decision table

Observed signal Safe next action Unsafe next action
Ledger missing exception rows Drive the strict=True path Rewrite error handling first
Tuple result stored as a list Keep canonicalize on compare Assert raw tuple == list
Bytes arg stored as a string Add the base64 type tag Call default=str and hope
One row fails after a rewrite Restore the body, then shrink Edit the JSONL to match
Many rows fail together The signature likely changed Continue with a larger patch
Results contain live objects Stop. Choose another seam Dump repr() as a fake oracle

Use the table when a model patch looks locally elegant.
Elegance is not equivalence. The ledger is equivalence.

Smallest safe change after the ledger is green

Allowed changes rewrite internals of that one function.
Local variable names inside that function may change.
Signature changes, default changes, and new kwargs stay forbidden.

Do not move the function into another module yet.
Do not extract a helper that callers might import.
Do not replace exception classes with cleaner looking types.

A candidate body can still be shorter and easier to read.
It must emit the same canonical results for every frozen row.
That is the entire patch. Stop after that patch is green.

# Allowed shape for a later body-only rewrite.
# Proposal only. Do not paste until the ledger is committed.

def normalize_ref(raw, strict=False):
    if raw is None:
        if strict:
            raise TypeError("ref required")
        return None
    if isinstance(raw, bytes):
        text = raw.decode("utf-8", errors="replace").strip()
    else:
        text = str(raw).strip()
    if text == "":
        if strict:
            raise ValueError("ref empty")
        return ""
    lowered = text.lower()
    if lowered in {"n/a", "na", "none", "null"}:
        return None
    if text.startswith("id:"):
        body = text[3:].strip()
        if body.isdigit():
            return ("id", int(body))
        if strict:
            raise ValueError("ref id not numeric")
        return ("id", None)
    if "," in text:
        return [p.strip() for p in text.split(",") if p.strip()] or ""
    return lowered
Enter fullscreen mode Exit fullscreen mode

Keep TypeError and ValueError on the original paths.
Keep the ("id", None) tuple on the non-strict invalid id.
Those details are the product. Formatting is not the product.

Using a free model only after the oracle exists

A model can propose a body rewrite against frozen tests.
It cannot invent the contract you failed to record.
Run the ledger tests on a server you control.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.
MonkeyCode offers free model access and a free server option.
Use that pair only after the JSONL oracle is committed.
Treat every model patch as a candidate, never as truth.
If you already have that free server, run these ledger tests there before merging the body rewrite.

Limitations

JSONL cannot store open sockets or live file handles.
Float equality needs a numeric tolerance, not raw equality.
Dict key order can hide semantically equal results if skipped.

Randomness, clocks, and network calls will break the ledger.
Freeze those sources first, or do not use this method.
Huge binary blobs will make the JSONL file unusable.

Message stems hide punctuation-only changes in error text.
That is deliberate. Full messages are usually too brittle.
Raise the stem length only when two errors collapse together.

Who should not use this

Do not use this on security-sensitive redaction paths.
Do not record secrets, tokens, or personal data.
Do not wrap functions that must stay allocation-free.

Teams without a test runner should not start here.
Write a seed script first, then add the recorder.
Skip this method if the function already has a formal spec.

Do not use this as a license for a wide rewrite.
One function. One body. One green ledger. Then stop.

Close

The refactor is the second commit, not the first.
The first commit is a canonical two-column ledger.
Change one function body and leave everything else alone.

Top comments (0)