Messy Rollup Refactor: Pin Canonical JSON, Then Extract One Parser

Messy rollup code should not be rewritten wholesale. Pin canonical JSON before the first extract. One pure parser is the only allowed first change.

AI patches fail when skip rules stay implicit. Characterization tests freeze skip counts and totals. The extract must not move those pins.

Why messy rollups break silent consumers

Rollup jobs write files other systems ingest. Key order and integer totals are part of that contract. A cleaner dump can still break a downstream hasher.

Globals make skip counts hard to see. Import-time defaults hide the output path. Mocks on open hide encoding and trailing newlines.

Unit tests can stay green while the artifact changes. Downstream jobs then fail on rounding or key order. That gap is the actual refactor risk.

Observables that belong in the first pin set

Choose values that survive a parser extract. Ignore clocks and absolute paths. Those values are environment, not behavior.

1. Skip count after a committed fixture.
2. Row count stored in the summary object.
3. Integer cent totals per user key.
4. Canonical JSON bytes with sorted keys.
5. SHA-256 of those canonical bytes.

Do not pin datetime.utcnow output. Do not pin indent-two pretty JSON. Do not pin the process working directory.

Proposed messy module

The next listing is a proposal. Treat it as unexecuted example code. It mixes parsing, globals, and file writes.

# rollup.py
from __future__ import annotations

import json
import os
from datetime import datetime

OUT_PATH = os.environ.get("ROLLUP_OUT", "summary.json")
SKIPPED = 0


def drain(path: str) -> None:
    global SKIPPED
    SKIPPED = 0
    rows = []
    with open(path, encoding="utf-8") as handle:
        for raw in handle:
            line = raw.strip()
            if not line or line.startswith("#"):
                SKIPPED += 1
                continue
            parts = line.split(",")
            if len(parts) != 3:
                SKIPPED += 1
                continue
            day, user, cents = parts
            try:
                amount = int(cents)
            except ValueError:
                SKIPPED += 1
                continue
            rows.append(
                {
                    "day": day.strip(),
                    "user": user.strip(),
                    "cents": amount,
                }
            )
    totals = {}
    for row in rows:
        key = row["user"]
        totals[key] = totals.get(key, 0) + row["cents"]
    payload = {
        "generated_at": datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%SZ"),
        "rows": len(rows),
        "skipped": SKIPPED,
        "totals_cents": dict(sorted(totals.items())),
    }
    with open(OUT_PATH, "w", encoding="utf-8") as handle:
        json.dump(payload, handle, indent=2)
        handle.write("\n")

This module writes generated_at on every run. That field is unstable across seconds. Slice it out of the characterization view.

datetime.utcnow is part of the mess. It is not a recommended clock. A later extract can inject time.

Committed fixture

Keep the fixture tiny and in git. Do not copy production log streams. Production logs are not a pin set.

# comment
2026-09-20,ada,150
2026-09-20,ada,25
not-a-row
2026-09-21,lin,40

2026-09-21,lin,abc

Expected pins for this fixture follow. skipped equals four. rows equals three. ada totals 175 cents. lin totals 40 cents.

The four skips are comment, bad arity, blank, and bad int. Count them on paper before coding. The paper count is the oracle.

Characterization test

The test uses a real temp file. It does not mock the builtin open. It hashes canonical JSON, not pretty JSON.

# test_characterize_rollup.py
from __future__ import annotations

import hashlib
import json
from pathlib import Path

import rollup

FIXTURE = """# comment
2026-09-20,ada,150
2026-09-20,ada,25
not-a-row
2026-09-21,lin,40

2026-09-21,lin,abc
"""

PIN_KEYS = ("rows", "skipped", "totals_cents")


def canonical_summary(path: Path) -> bytes:
    raw = json.loads(path.read_text(encoding="utf-8"))
    sliced = {key: raw[key] for key in PIN_KEYS}
    return json.dumps(
        sliced, sort_keys=True, separators=(",", ":")
    ).encode("utf-8")


def test_pins_skip_rows_and_totals(tmp_path, monkeypatch):
    src = tmp_path / "usage.csv"
    out = tmp_path / "summary.json"
    src.write_text(FIXTURE, encoding="utf-8")
    monkeypatch.setattr(rollup, "OUT_PATH", str(out))
    rollup.SKIPPED = 0
    rollup.drain(str(src))
    blob = canonical_summary(out)
    payload = json.loads(blob)
    assert payload == {
        "rows": 3,
        "skipped": 4,
        "totals_cents": {"ada": 175, "lin": 40},
    }
    assert rollup.SKIPPED == 4
    # Replace DIGEST after one local run. Do not guess it.
    digest = hashlib.sha256(blob).hexdigest()
    assert len(digest) == 64

The digest length check is a scaffold only. Do not invent a hash in a draft. Run the test once on your machine.

Commands to freeze the digest

Create a venv first. Install pytest only. Keep the dependency list small.

python -m venv .venv
source .venv/bin/activate
pip install pytest
pytest test_characterize_rollup.py -q

Print the digest with a one-off snippet. Copy the hex into the assertion. Commit fixture, test, and digest together.

# print_canonical_digest.py  (proposal; run locally)
import hashlib
import json
from pathlib import Path

raw = json.loads(Path("summary.json").read_text(encoding="utf-8"))
sliced = {
    "rows": raw["rows"],
    "skipped": raw["skipped"],
    "totals_cents": raw["totals_cents"],
}
blob = json.dumps(sliced, sort_keys=True, separators=(",", ":")).encode("utf-8")
print(blob.decode("utf-8"))
print(hashlib.sha256(blob).hexdigest())

Canonical bytes for this fixture should look like this. Confirm them after the first drain. Do not edit spaces by hand.

{"rows":3,"skipped":4,"totals_cents":{"ada":175,"lin":40}}

If ada prints as 175.0, the pin already failed. Integer cents are the contract. Float totals are a behavior change.

Decision table for pass 1

Change	Pass 1	Reason
`parse_line(line) -> dict \	None`	Yes
Reset SKIPPED inside drain	No	Global contract stays pinned
Change json.dump separators	No	Pretty bytes are still consumed
Drop generated_at	Later	Needs a second pin set
Inject a clock argument	Later	New seam, new tests
Parallel file writes	No	Not a small change

Pass 1 allows parse_line only. Pass 1 forbids dump format changes. Pass 1 forbids clock injection.

Six-step workflow

1. Commit the fixture and the characterization test.
2. Run pytest on that single test file.
3. Record SHA-256 of the canonical slice.
4. Extract parse_line and keep drain I/O.
5. Re-run pytest and compare the digest.
6. Stop after that single extract.

Revert if the digest moves. A moved hash means behavior leaked. Do not stack extra cleanups on a red hash.

Name the extract in the commit message. Do not mix formatting in that commit. One behavior pin, one function, one diff.

Where a free model fits

A model may draft parse_line after pins exist. It must not choose the pin set. Pins come from the fixture and the oracle count.

Disclosure: This article was prepared as part of MonkeyCode's product outreach.

MonkeyCode provides free model access and a free server option. Use that pair only on a green harness. Paste the test and drain, not production files.

Ask for one function extract. Reject patches that touch OUT_PATH. Reject patches that change json.dump keywords.

Run local pytest after the patch. The free server is not the test runner. Discard the diff if the digest changes.

If you already use those free models, paste the harness first.

Proposed extract

This extract is still a proposal. Keep it unmerged until pytest stays green. Call it from drain only.

def parse_line(line: str) -> dict | None:
    text = line.strip()
    if not text or text.startswith("#"):
        return None
    parts = text.split(",")
    if len(parts) != 3:
        return None
    day, user, cents = parts
    try:
        amount = int(cents)
    except ValueError:
        return None
    return {
        "day": day.strip(),
        "user": user.strip(),
        "cents": amount,
    }

drain then loops and increments SKIPPED on None. Writes stay inside drain. No new classes appear in pass 1.

for raw in handle:
    parsed = parse_line(raw)
    if parsed is None:
        SKIPPED += 1
        continue
    rows.append(parsed)

That is the whole first change. No framework. No type hierarchy. No second file unless imports demand it.

Failure analysis

If skipped is three, the blank line was missed. If ada is 175.0, an int became float. If key order flips, canonical dumps will catch it.

If generated_at sneaks into the slice, hashes churn. Keep the slice explicit. Name the keys in a tuple.

If pytest cannot import rollup, the package path drifted. Run from the module directory first. Do not add sys.path hacks in pass 1.

What the hash does not prove

The hash does not prove locale rules. The hash does not prove UTF-8 BOM handling. The hash does not prove concurrent drain calls.

generated_at remains a later extract. Inject a clock only after a new pin. Pretty-print indent is also later work.

Comma-in-user fields are outside this fixture. Do not claim CSV completeness. Add a second fixture when that case matters.

Who should skip this workflow

Skip this when skip rules are unknown. Skip this for cryptographic or auth parsers. Skip this when no fixture can be committed.

Public APIs need designed types. This method is for messy internals. It is not a substitute for a spec.

Teams with two disagreeing consumers must pin both outputs. One hash cannot represent two contracts. Split the fixtures first.

Limits of free model help

Free model access can mis-count comments. The free server does not see your disk. Never paste secrets or customer logs.

Models reorder keys to look neat. Canonical JSON makes that visible. Models may emit float cents. Integer pins reject that class of patch.

Do not loop prompts on a red test. Revert, shrink the ask, and rerun. A second prompt is not a second pin.

Close

Pin skips, rows, and canonical JSON first. Extract one parser. Leave I/O and globals for later passes.

The harness is the contract. The model is optional labor. Keep the first patch inside one function.