Characterize the Job Envelope Before You Touch Helpers

Characterization must precede every cleanup in a messy repo. Skip the envelope and the refactor becomes a guess. Capture observable side effects before any internal edit.

This article shows a two-stage workflow for brownfield jobs. Stage one records a behavior envelope on disk. Stage two allows one helper change against that envelope.

The method fits scripts with mixed I/O and weak tests. It is not a substitute for later unit coverage.

What an envelope actually records

A behavior envelope is a frozen observation of job I/O. It is not a unit test of internal helpers. It is a contract around the process edge.

Record these seven fields for every fixture run.

1. Record the argument vector after the interpreter name.
2. Record selected environment keys with stable values only.
3. Store canonical hashes of every declared input file.
4. Store standard output bytes encoded as base64 text.
5. Store standard error bytes encoded as base64 text.
6. Store the process exit code as an integer.
7. Store canonical hashes of every declared output file.

Hash the canonical JSON of those seven fields. That digest is the only merge gate used. Any unapproved digest change fails the refactor review.

Why helpers are the wrong first target

Messy repos hide behavior in helpers, globals, and file writes. Editing a helper without an envelope deletes evidence. Reviewers then argue about style instead of drift.

Fan-in checks and public-surface diffs still matter later. They do not replace a job-level oracle. Start the refactor at the process edge.

Synthetic job under test

The fixture below is a proposal, not production code. It mixes parsing, logging, and file output on purpose. Treat it as a stand-in for a brownfield batch script.

# messy_job.py — synthetic fixture, not a live service
from __future__ import annotations

import json
import os
import sys
from pathlib import Path


def _load_rates(path: Path) -> dict:
    raw = json.loads(path.read_text(encoding="utf-8"))
    return {str(k): float(v) for k, v in raw.items()}


def _apply_markup(amount: float, rates: dict, region: str) -> float:
    rate = rates.get(region, rates.get("DEFAULT", 0.0))
    # Hidden branch: empty region reuses DEFAULT twice.
    if region == "":
        rate = rates.get("DEFAULT", 0.0)
    return round(amount * (1.0 + rate), 2)


def _write_report(path: Path, rows: list[dict]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8")


def main(argv: list[str] | None = None) -> int:
    argv = list(sys.argv[1:] if argv is None else argv)
    if len(argv) < 3:
        sys.stderr.write("usage: messy_job.py RATES IN OUT\n")
        return 2
    region = os.environ.get("JOB_REGION", "DEFAULT")
    rates = _load_rates(Path(argv[0]))
    incoming = json.loads(Path(argv[1]).read_text(encoding="utf-8"))
    rows = []
    for item in incoming:
        amount = float(item["amount"])
        tagged = _apply_markup(amount, rates, region)
        rows.append({"id": item["id"], "total": tagged})
        sys.stdout.write(f"{item['id']} {tagged}\n")
    _write_report(Path(argv[2]), rows)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Three coupling points make a naive cleanup unsafe. Environment selection changes the numeric region path. Stdout and the output file both carry totals.

Fixture files used by the harness

Keep fixtures tiny, named, and legally copyable. Do not pull live customer files into git. Two region values are enough for this job.

{
  "DEFAULT": 0.05,
  "EU": 0.08
}

[
  {"id": "a1", "amount": 100},
  {"id": "a2", "amount": 2.5}
]

{
  "name": "region_empty",
  "job": "messy_job.py",
  "files": {
    "rates": "fixtures/rates.json",
    "incoming": "fixtures/incoming.json",
    "outgoing": "outgoing.json"
  }
}

Export one region value per fixture before recording. Do not record both regions in a single process. Mixed region state will poison the digest.

Artifact: envelope recorder

The recorder below is a local harness. Run it against the synthetic job only. It writes one JSON envelope per fixture name.

# envelope_record.py — proposal harness for the synthetic job
from __future__ import annotations

import base64
import hashlib
import json
import os
import subprocess
import sys
from pathlib import Path
from typing import Any

ENV_ALLOWLIST = ("JOB_REGION",)
DIGEST_KEYS = (
    "env",
    "exit_code",
    "inputs",
    "outputs",
    "stderr_b64",
    "stdout_b64",
)


def sha256_file(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as handle:
        for chunk in iter(lambda: handle.read(65536), b""):
            digest.update(chunk)
    return digest.hexdigest()


def canonical_dumps(payload: dict[str, Any]) -> str:
    return json.dumps(
        payload, sort_keys=True, separators=(",", ":"), ensure_ascii=True
    )


def digest_of(envelope: dict[str, Any]) -> str:
    body = {key: envelope[key] for key in DIGEST_KEYS}
    return hashlib.sha256(canonical_dumps(body).encode("utf-8")).hexdigest()


def record(manifest_path: Path, out_path: Path) -> None:
    manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    work = Path("envelope_work") / manifest["name"]
    if work.exists():
        raise SystemExit(f"refuse to reuse {work}")
    work.mkdir(parents=True)

    names = {
        "rates": Path(manifest["files"]["rates"]).name,
        "incoming": Path(manifest["files"]["incoming"]).name,
        "outgoing": Path(manifest["files"]["outgoing"]).name,
    }
    (work / names["rates"]).write_bytes(
        Path(manifest["files"]["rates"]).read_bytes()
    )
    (work / names["incoming"]).write_bytes(
        Path(manifest["files"]["incoming"]).read_bytes()
    )
    argv = [names["rates"], names["incoming"], names["outgoing"]]

    recorded_env = {
        key: os.environ[key] for key in ENV_ALLOWLIST if key in os.environ
    }
    run_env = {"PATH": os.environ.get("PATH", "")}
    run_env.update(recorded_env)

    proc = subprocess.run(
        [sys.executable, str(Path(manifest["job"]).resolve()), *argv],
        cwd=str(work),
        env=run_env,
        capture_output=True,
        check=False,
    )
    outgoing = work / names["outgoing"]
    outputs = {}
    if outgoing.exists():
        outputs["outgoing"] = sha256_file(outgoing)

    envelope = {
        "name": manifest["name"],
        "argv": argv,
        "env": recorded_env,
        "inputs": {
            "rates": sha256_file(work / names["rates"]),
            "incoming": sha256_file(work / names["incoming"]),
        },
        "stdout_b64": base64.b64encode(proc.stdout).decode("ascii"),
        "stderr_b64": base64.b64encode(proc.stderr).decode("ascii"),
        "exit_code": proc.returncode,
        "outputs": outputs,
    }
    envelope["digest"] = digest_of(envelope)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(envelope, indent=2) + "\n", encoding="utf-8")
    sys.stdout.write(f"{manifest['name']} {envelope['digest']}\n")


if __name__ == "__main__":
    if len(sys.argv) != 3:
        raise SystemExit("usage: envelope_record.py MANIFEST OUT")
    record(Path(sys.argv[1]), Path(sys.argv[2]))

The recorder copies inputs into a sandbox directory. It then executes the job with a trimmed environment. Output files are hashed after the process exits.

This harness assumes a Unix-like PATH for the interpreter. Windows users must extend the inherited environment map. Missing PATH values will fail before the job starts.

Artifact: envelope checker

The checker loads a golden envelope from disk. It replays the same argv, env, and inputs. It fails when the canonical digest does not match.

# envelope_check.py — proposal harness for the synthetic job
from __future__ import annotations

import json
import os
import shutil
import sys
from pathlib import Path

from envelope_record import digest_of, record


def check(manifest_path: Path, golden_path: Path) -> int:
    golden = json.loads(golden_path.read_text(encoding="utf-8"))
    work = Path("envelope_work") / golden["name"]
    if work.exists():
        shutil.rmtree(work)
    tmp = Path("envelopes") / "_replay.json"
    os.environ.update(golden["env"])
    record(manifest_path, tmp)
    replay = json.loads(tmp.read_text(encoding="utf-8"))
    expected = golden["digest"]
    actual = digest_of(replay)
    if actual != expected:
        sys.stderr.write(f"digest mismatch {expected} != {actual}\n")
        return 1
    sys.stdout.write(f"ok {golden['name']} {actual}\n")
    return 0


if __name__ == "__main__":
    if len(sys.argv) != 3:
        raise SystemExit("usage: envelope_check.py MANIFEST GOLDEN")
    raise SystemExit(check(Path(sys.argv[1]), Path(sys.argv[2])))

Do not parse the JSON by eye during review. Compare the digest field only at first. Open the full envelope after a mismatch.

export JOB_REGION=""
python envelope_record.py fixtures/region_empty.json envelopes/region_empty.json
export JOB_REGION="DEFAULT"
python envelope_record.py fixtures/region_default.json envelopes/region_default.json
python envelope_check.py fixtures/region_empty.json envelopes/region_empty.json

Commit golden envelopes before any helper patch. Treat those files as the oracle. A later digest move needs an explicit human allow.

Workflow

Follow this sequence on a throwaway branch.

1. Collect two real fixtures with known business meaning.
2. Declare input paths and output paths in a manifest.
3. Run the recorder and commit envelopes as golden files.
4. Delete any helper edit that landed before step three.
5. Change one helper, then rerun the checker.
6. If the digest moves, revert the helper immediately.
7. If the digest holds, keep the helper change only.
8. Add a unit test only after the envelope stays green.

Two fixtures beat one for hidden region branches. Empty region and DEFAULT often diverge in messy jobs. Encode both in the manifest before any cleanup.

Decision table

Use this table when a check fails. It maps a symptom to the allowed next edit.

Symptom	Envelope action	Allowed next edit
Digest matches after a rename	Keep golden files	Rename only
Stdout differs, files match	Need product sign-off	Do not touch helpers yet
Output file hash differs	Treat as behavior change	Stop the refactor
Exit code differs	Treat as behavior change	Stop the refactor
Env key leaked into the job	Shrink the env allowlist	No helper edit
Clock or UUID in output	Strip that field from capture	No helper edit
Network call inside the job	Pin a fixture response file	No helper edit

Product sign-off means a human owner of the job. A model suggestion is not product sign-off. A green unit test is not sign-off either.

Smallest safe change

The change below is a proposal on the synthetic job. It extracts rounding and does not alter totals. Rerun the checker immediately after this edit.

def _round_money(value: float) -> float:
    return round(value, 2)


def _apply_markup(amount: float, rates: dict, region: str) -> float:
    rate = rates.get(region, rates.get("DEFAULT", 0.0))
    if region == "":
        rate = rates.get("DEFAULT", 0.0)
    return _round_money(amount * (1.0 + rate))

Stop the branch after this single extraction. Do not relocate the loader in the same commit. Do not rewrite logging in the same commit.

A safe commit touches one helper and nothing else. If the checker is silent, the envelope still holds. Silence is the only success signal here.

Where a free model belongs

MonkeyCode offers free model access and a free server option. Disclosure: This article was prepared as part of MonkeyCode's product outreach. A free coding model can draft recorder boilerplate from the field list.

Use those only to draft the harness, not to bless the digest. Keep the golden envelopes inside your own repository. Copy any remote run artifacts back before the first helper edit.

The model does not own the seven-field schema. The server does not own the merge gate. Human reviewers keep ownership of both controls.

Limitations

The envelope ignores internal call graphs on purpose. It will not catch a pure rename that still compiles. It will not catch a leaked secret in new logs.

Clocks, randomness, and network I/O break the digest. Strip those or pin them before recording. Do not hash live timestamps and expect stability.

Large binary outputs will bloat the git repository. Store hashes for binaries, not the bytes. Keep byte capture for small text streams only.

Parallel jobs that race on the same output path will flake. Serialize those jobs during the characterization window. Flakes are envelope bugs, not helper bugs.

Who should skip this

Skip this workflow if the job has no replayable fixtures. Skip it if writes cannot be sandboxed on disk. Skip it if legal rules block copying production-like inputs.

Do not use this as a production deploy gate alone. Do not use this to justify a thousand-line cleanup. Do not use this when a real contract test suite already exists.

Teams with strong typed boundaries may not need envelopes. They still need some oracle before AI-shaped diffs land. An envelope is one oracle, not the only oracle.

Close

Lock the job envelope before any helper cleanup. Then change one function and rerun the checker. If a free server run isolates the job, copy envelopes back first.