At Level 5 Operator, a successful calculation is not enough. You must preserve the source, code, checks, comparison, decision, and published value so another operator can reproduce the claim after its builder has left. This book supplies one deliberately small, standard-library implementation for both a Lab figure and a Company board-pack number. Every record is synthetic.
2. A green run becomes a company-wide incident
At 08:40 you approve a monthly result. At 09:00 it appears in a board pack; at 09:15 regional managers act on it. Then an analyst says the CRM export was replaced after the pipeline ran. The run page is green, but it links to latest.csv, old outputs were overwritten, and nobody exported the scheduler parameters. You cannot prove whether the displayed 947 came from the reviewed source or who authorised that exact value.
You stop publication while Finance and Operations wait. This is the Level 5 blast radius: decisions can be wrong across the organisation, and recovery cannot depend on the builder's memory. A digest proves byte identity, not truth, authority, privacy, or lawful retention. Those controls must remain explicit.
3. After this you can
- Seal source, output, code, schema, checks, runtime, and parameters under an immutable run ID.
- Compare a candidate with one identified, approved baseline.
- Bind a separate preparer's request and approver's decision to exact digests and a destination.
- Trace a Lab figure or Company board number back to contributing synthetic rows.
- Assign retention and deletion duties without treating “keep everything” as provenance.
4. Prerequisites
T10-L04- Pipelines over real data, especially read-only SQLite, exact schema gates, safe-field selection, and atomic output.T12-L05- Governance, evidence and handover, especially ownership, evidence, retention, restore, and the author-has-left test.- Python 3.10 or newer and a disposable empty directory.
- Two exercise identities:
operator-aandapprover-b. Real permissions must enforce that separation.
Use only public, synthetic, course-provided, or explicitly approved data. Never substitute customer, employee, participant, unpublished, credential, or unrestricted trace data. Before adapting this exercise, confirm ownership, classification, purpose, approved storage, readers, geography, retention, backup expiry, and deletion verification.
5. The idea in one page
A published number is a claim with a chain of evidence:
publication receipt and displayed value
-> approval request and its digest
-> sealed manifest and output digest
-> quality checks and baseline drift report
-> exact schema, parameters, code, and runtime
-> immutable source snapshot and its digest
Use unique run IDs, never latest. Test format drift first: changed names, types, keys, units, encodings, or timestamp conventions stop the calculation. Then test distribution drift against a named approved baseline. Thresholds are investigation boundaries chosen before seeing a candidate, not proof that data inside them is correct. Segment checks matter because a stable total can hide a missing region or instrument.
Acceptance is not publication. A preparer identifies one run, metric, rendered value, and destination. A different approver recomputes the bound digests before writing a receipt. Production storage must enforce append-only or object-lock behaviour; a JSON field cannot make writable storage immutable.
Retention is object-specific. Raw snapshots, rejected candidates, logs, approved evidence, requests, and receipts can have different purposes and periods. Record owner, trigger, due date, legal hold, replicas, backups, executor, and independent verifier. Deleting a primary file does not delete its backup.
6. Run the complete Lab and Company paths
The earlier draft described several scripts but did not provide a closed implementation. This final pass deliberately supports one file instead: steps/operate.py. Do not also create partial seal_run.py or approve.py variants. The script below generates fixtures, applies exact schemas and safe-field queries, calculates metrics, seals baselines and candidates atomically, compares drift, creates requests, and independently decides them. It uses only Python's standard library.
Create an empty project, make a steps directory, and save this complete file as steps/operate.py:
import argparse, hashlib, json, os, re, shutil, sqlite3, sys
from datetime import datetime, timezone
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
BASE_IDS = {"lab": "LAB-BASE-001", "company": "CRM-BASE-001"}
DESTINATIONS = {
"lab": "synthetic manuscript Figure 2 caption",
"company": "synthetic August board pack slide 4 cell B7",
}
SCHEMAS = {
"lab": [["measurement_id","TEXT",1,1], ["batch_code","TEXT",1,0],
["measured_at","TEXT",1,0], ["value_c","REAL",1,0],
["operator_note","TEXT",1,0]],
"company": [["activity_id","TEXT",1,1], ["region","TEXT",1,0],
["activity_type","TEXT",1,0], ["occurred_at","TEXT",1,0],
["qualified","INTEGER",1,0], ["contact_email","TEXT",1,0]],
}
def canonical(value):
return json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False).encode()
def digest_bytes(data): return hashlib.sha256(data).hexdigest()
def digest_file(path): return digest_bytes(Path(path).read_bytes())
def write_json(path, value):
path = Path(path); path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(canonical(value) + b"\n")
def read_json(path): return json.loads(Path(path).read_text(encoding="utf-8"))
def now(): return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
def db_path(lane): return ROOT / "source" / ("lab.db" if lane == "lab" else "crm.db")
def work_path(lane): return ROOT / "work" / f"{lane}.json"
def fixture(lane):
path = db_path(lane); path.parent.mkdir(parents=True, exist_ok=True)
path.unlink(missing_ok=True)
with sqlite3.connect(path) as con:
if lane == "lab":
con.execute("CREATE TABLE measurements (measurement_id TEXT PRIMARY KEY NOT NULL, batch_code TEXT NOT NULL, measured_at TEXT NOT NULL, value_c REAL NOT NULL, operator_note TEXT NOT NULL)")
rows = [("M-101","CEDAR-A","2026-09-01T09:00:00+00:00",18.2,"synthetic"),
("M-102","CEDAR-A","2026-09-01T09:05:00+00:00",19.1,"synthetic"),
("M-103","CEDAR-B","2026-09-01T09:10:00+00:00",20.0,"synthetic"),
("M-104","CEDAR-B","2026-09-01T09:15:00+00:00",21.4,"synthetic")]
con.executemany("INSERT INTO measurements VALUES (?,?,?,?,?)", rows)
else:
con.execute("CREATE TABLE activities (activity_id TEXT PRIMARY KEY NOT NULL, region TEXT NOT NULL, activity_type TEXT NOT NULL, occurred_at TEXT NOT NULL, qualified INTEGER NOT NULL, contact_email TEXT NOT NULL)")
rows = [(f"A-{n:04d}", ["NORTH","SOUTH","WEST"][(n-1)%3],
["CALL","MEETING"][(n-1)%2],
f"2026-08-{((n-1)%28)+1:02d}T09:00:00+00:00",
1 if n <= 947 else 0, f"fictional-{n:04d}@example.invalid")
for n in range(1, 1001)]
con.executemany("INSERT INTO activities VALUES (?,?,?,?,?,?)", rows)
print(f"FIXTURE_READY lane={lane} sha256={digest_file(path)}")
def schema_of(con, table):
return [[r[1], r[2], r[3], r[5]] for r in con.execute(f"PRAGMA table_info({table})")]
def build(lane):
path = db_path(lane)
if not path.exists(): raise SystemExit("fixture missing")
uri = path.resolve().as_uri() + "?mode=ro"
with sqlite3.connect(uri, uri=True) as con:
con.execute("PRAGMA query_only=ON")
table = "measurements" if lane == "lab" else "activities"
observed_schema = schema_of(con, table)
if observed_schema != SCHEMAS[lane]: raise SystemExit("exact schema gate failed")
if lane == "lab":
rows = con.execute("SELECT measurement_id,batch_code,measured_at,value_c FROM measurements ORDER BY measurement_id").fetchall()
groups = {g: [r[3] for r in rows if r[1] == g] for g in sorted({r[1] for r in rows})}
metrics = {"measurement_count": len(rows),
"mean_value_c": sum(r[3] for r in rows)/len(rows),
"batch_mean_value_c": {g: sum(v)/len(v) for g,v in groups.items()}}
results = [check("schema-exact", True, digest_bytes(canonical(observed_schema))),
check("row-count-4-to-6", 4 <= len(rows) <= 6, len(rows)),
check("unique-id", len({r[0] for r in rows}) == len(rows), len({r[0] for r in rows})),
check("timestamp-offset", all(r[2].endswith("+00:00") for r in rows), len(rows)),
check("value-range-0-to-50", all(0 <= r[3] <= 50 for r in rows), [min(r[3] for r in rows),max(r[3] for r in rows)])]
safe_rows = [list(r) for r in rows]
else:
rows = con.execute("SELECT activity_id,region,activity_type,occurred_at,qualified FROM activities ORDER BY activity_id").fetchall()
if any(r[4] not in (0,1) for r in rows): raise SystemExit("qualified-domain gate failed")
qualified = [r for r in rows if r[4] == 1]
regions, kinds = sorted({r[1] for r in rows}), sorted({r[2] for r in rows})
by_region = {g: sum(r[1] == g for r in qualified) for g in regions}
by_kind = {g: sum(r[2] == g for r in qualified) for g in kinds}
metrics = {"activity_count": len(rows), "qualified_activity_count": len(qualified),
"qualified_by_region": by_region, "qualified_by_activity_type": by_kind}
results = [check("schema-exact", True, digest_bytes(canonical(observed_schema))),
check("row-count-1000", len(rows) == 1000, len(rows)),
check("qualified-domain", all(r[4] in (0,1) for r in rows), len(qualified)),
check("timestamp-offset", all(r[3].endswith("+00:00") for r in rows), len(rows)),
check("privacy-field-absence", all(len(r) == 5 for r in rows), "contact_email not selected"),
check("region-reconciliation", sum(by_region.values()) == len(qualified), sum(by_region.values())),
check("type-reconciliation", sum(by_kind.values()) == len(qualified), sum(by_kind.values()))]
safe_rows = [list(r) for r in rows]
if any(r["result"] != "PASS" for r in results): raise SystemExit("quality gate failed")
document = {"lane": lane, "source_snapshot_id": "SYN-LAB-2026-09-R1" if lane == "lab" else "SYN-CRM-2026-08-R1",
"source_sha256": digest_file(path), "schema": observed_schema,
"schema_fingerprint": digest_bytes(canonical(observed_schema)),
"metrics": metrics, "safe_rows": safe_rows,
"checks": {"check_set_version": 1, "results": results, "overall": "PASS"}}
write_json(work_path(lane), document)
print(f"BUILD_PASS lane={lane} output={work_path(lane)}")
def check(name, passed, observed):
return {"check_id": name, "observed": observed, "result": "PASS" if passed else "FAIL"}
def metric(document, path):
value = document
for part in path.split("."): value = value[part]
return value
def verified_baseline(lane, run_id):
state_path = ROOT / "state" / f"{lane}-baseline.json"
if not state_path.exists(): raise SystemExit("approved baseline state missing")
state = read_json(state_path)
if state["run_id"] != run_id: raise SystemExit("baseline differs from approved state")
receipt = ROOT / state["receipt_file"]
if digest_file(receipt) != state["receipt_sha256"]: raise SystemExit("baseline receipt changed")
manifest = ROOT / "runs" / run_id / "manifest.json"
if digest_file(manifest) != state["manifest_sha256"]: raise SystemExit("baseline manifest changed")
return read_json(manifest), read_json(ROOT / "runs" / run_id / "output.json")
def drift(lane, baseline, candidate, baseline_manifest, run_id):
bm, cm = baseline["metrics"], candidate["metrics"]
if lane == "lab":
checks = [delta("row-change", bm["measurement_count"], cm["measurement_count"], .25),
absolute("mean-change-c", bm["mean_value_c"], cm["mean_value_c"], 2.0),
groups("new-batch-code", bm["batch_mean_value_c"], cm["batch_mean_value_c"])]
else:
checks = [delta("row-change", bm["activity_count"], cm["activity_count"], .15),
delta("qualified-count-change", bm["qualified_activity_count"], cm["qualified_activity_count"], .15),
groups("new-region", bm["qualified_by_region"], cm["qualified_by_region"]),
groups("new-activity-type", bm["qualified_by_activity_type"], cm["qualified_by_activity_type"])]
return {"report_version": 1, "baseline_run_id": baseline_manifest["run_id"],
"candidate_run_id": run_id, "baseline_manifest_sha256": digest_bytes(canonical(baseline_manifest)+b"\n"),
"checks": checks, "overall": "PASS" if all(c["result"] == "PASS" for c in checks) else "FAIL"}
def delta(name, old, new, threshold):
observed = abs(new-old)/max(abs(old),1)
return {"check_id": name, "observed": observed, "threshold": threshold, "result": "PASS" if observed <= threshold else "FAIL"}
def absolute(name, old, new, threshold):
observed = abs(new-old)
return {"check_id": name, "observed": observed, "threshold": threshold, "result": "PASS" if observed <= threshold else "FAIL"}
def groups(name, old, new):
observed = sorted(set(new)-set(old))
return {"check_id": name, "observed": observed, "threshold": [], "result": "PASS" if not observed else "FAIL"}
def seal(lane, run_id, bootstrap, baseline_id):
if not re.fullmatch(r"[A-Z0-9-]{6,40}", run_id): raise SystemExit("unsafe run id")
final, stage = ROOT/"runs"/run_id, ROOT/"runs"/f".staging-{run_id}"
if final.exists() or stage.exists(): raise SystemExit("run id already exists")
document = read_json(work_path(lane))
if document["lane"] != lane or document["source_sha256"] != digest_file(db_path(lane)):
raise SystemExit("work output does not match current source")
if document["checks"]["overall"] != "PASS": raise SystemExit("failed checks")
if bootstrap:
if baseline_id or run_id != BASE_IDS[lane] or (ROOT/"state"/f"{lane}-baseline.json").exists():
raise SystemExit("bootstrap policy rejected")
report = {"status": "NOT_APPLICABLE", "reason": "FIRST_APPROVED_BASELINE"}
status = "BASELINE_CANDIDATE"
else:
if not baseline_id: raise SystemExit("candidate requires --baseline")
base_manifest, base_output = verified_baseline(lane, baseline_id)
report = drift(lane, base_output, document, base_manifest, run_id)
if report["overall"] != "PASS": raise SystemExit("drift failed")
status = "CANDIDATE"
stage.mkdir(parents=True)
try:
shutil.copy2(db_path(lane), stage/"input.db")
write_json(stage/"output.json", document)
write_json(stage/"check-results.json", document["checks"])
write_json(stage/"drift-report.json", report)
files = {p.name: {"sha256": digest_file(p), "bytes": p.stat().st_size} for p in stage.iterdir()}
script_hash = digest_file(__file__)
manifest = {"manifest_version": 2, "run_id": run_id, "lane": lane, "created_at": now(),
"source": {"snapshot_id": document["source_snapshot_id"], "file": "input.db", "sha256": files["input.db"]["sha256"]},
"code": {"file": "steps/operate.py", "sha256": script_hash},
"parameters": {"implementation": "operate-v1", "sha256": digest_bytes(canonical({"lane":lane,"schema":SCHEMAS[lane]}))},
"runtime": {"python": sys.version.split()[0], "dependency_mode": "standard-library-only"},
"schema": {"canonicalisation": "compact-sorted-json-utf8-v1", "fingerprint_sha256": document["schema_fingerprint"]},
"quality": {"file": "check-results.json", "overall": "PASS", "sha256": files["check-results.json"]["sha256"]},
"drift": report, "output": {"file": "output.json", "sha256": files["output.json"]["sha256"]},
"retention": {"policy_id": "synthetic-training-v1", "class": "approved-evidence", "delete_due": "2027-09-04", "legal_hold": False},
"publication_status": status, "files": files}
write_json(stage/"manifest.json", manifest)
stage.replace(final)
except Exception:
shutil.rmtree(stage, ignore_errors=True); raise
print(f"SEALED run={run_id} manifest_sha256={digest_file(final/'manifest.json')}")
def request(run_id, request_id, purpose, prepared, approver, metric_path, rendered, destination):
if prepared == approver: raise SystemExit("self-approval forbidden")
run = ROOT/"runs"/run_id; manifest = read_json(run/"manifest.json")
if purpose == "DRIFT_BASELINE":
if manifest["publication_status"] != "BASELINE_CANDIDATE": raise SystemExit("not a baseline candidate")
metric_path = rendered = destination = None
else:
if manifest["publication_status"] != "CANDIDATE" or destination != DESTINATIONS[manifest["lane"]]:
raise SystemExit("candidate or destination rejected")
if not metric_path or rendered is None: raise SystemExit("metric and rendering required")
output = read_json(run/"output.json")
body = {"request_version": 1, "request_id": request_id, "purpose": purpose,
"run_id": run_id, "manifest_sha256": digest_file(run/"manifest.json"),
"output_sha256": digest_file(run/"output.json"), "metric_path": metric_path,
"stored_value": metric(output, metric_path) if metric_path else None,
"rendered_value": rendered, "destination": destination,
"prepared_by": prepared, "requested_approver": approver, "requested_at": now()}
path = ROOT/"publications"/"requests"/f"{request_id}.json"
if path.exists(): raise SystemExit("request id exists")
write_json(path, body); print(f"REQUESTED file={path}")
def decide(request_file, decision, actor):
path = Path(request_file).resolve(); request_body = read_json(path)
if decision != "APPROVED" or actor != request_body["requested_approver"] or actor == request_body["prepared_by"]:
raise SystemExit("approval separation or decision rejected")
run = ROOT/"runs"/request_body["run_id"]; manifest_path, output_path = run/"manifest.json", run/"output.json"
if digest_file(manifest_path) != request_body["manifest_sha256"] or digest_file(output_path) != request_body["output_sha256"]:
raise SystemExit("sealed bytes changed")
manifest, output = read_json(manifest_path), read_json(output_path)
if manifest["quality"]["overall"] != "PASS": raise SystemExit("quality failed")
if request_body["purpose"] == "PUBLICATION":
if manifest["drift"]["overall"] != "PASS" or metric(output, request_body["metric_path"]) != request_body["stored_value"]:
raise SystemExit("drift or metric verification failed")
receipt = {"receipt_version": 1, "decision": decision, "purpose": request_body["purpose"],
"request_id": request_body["request_id"], "request_sha256": digest_file(path),
"run_id": request_body["run_id"], "manifest_sha256": request_body["manifest_sha256"],
"output_sha256": request_body["output_sha256"], "metric_path": request_body["metric_path"],
"stored_value": request_body["stored_value"], "rendered_value": request_body["rendered_value"],
"destination": request_body["destination"], "prepared_by": request_body["prepared_by"],
"approved_by": actor, "approved_at": now(),
"retention": {"policy_id":"synthetic-training-v1","delete_due":"2027-09-04"}}
receipt_path = ROOT/"publications"/"receipts"/f"{request_body['request_id']}.json"
if receipt_path.exists(): raise SystemExit("receipt already exists")
write_json(receipt_path, receipt)
if request_body["purpose"] == "DRIFT_BASELINE":
state_path = ROOT/"state"/f"{manifest['lane']}-baseline.json"
if state_path.exists(): raise SystemExit("active baseline already exists")
relative = receipt_path.relative_to(ROOT).as_posix()
write_json(state_path, {"run_id": request_body["run_id"], "manifest_sha256": request_body["manifest_sha256"],
"receipt_file": relative, "receipt_sha256": digest_file(receipt_path)})
print(f"APPROVED receipt={receipt_path}")
parser = argparse.ArgumentParser()
sub = parser.add_subparsers(dest="command", required=True)
for name in ("fixture","build"):
p=sub.add_parser(name); p.add_argument("--lane", choices=BASE_IDS, required=True)
p=sub.add_parser("seal"); p.add_argument("--lane", choices=BASE_IDS, required=True); p.add_argument("--run-id", required=True)
p.add_argument("--bootstrap-baseline", action="store_true"); p.add_argument("--baseline")
p=sub.add_parser("request"); p.add_argument("--run-id", required=True); p.add_argument("--request-id", required=True)
p.add_argument("--purpose", choices=["DRIFT_BASELINE","PUBLICATION"], required=True); p.add_argument("--prepared-by", required=True)
p.add_argument("--requested-approver", required=True); p.add_argument("--metric-path"); p.add_argument("--rendered-value"); p.add_argument("--destination")
p=sub.add_parser("decide"); p.add_argument("--request", required=True); p.add_argument("--decision", required=True); p.add_argument("--actor", required=True)
a=parser.parse_args()
if a.command == "fixture": fixture(a.lane)
elif a.command == "build": build(a.lane)
elif a.command == "seal": seal(a.lane, a.run_id, a.bootstrap_baseline, a.baseline)
elif a.command == "request": request(a.run_id,a.request_id,a.purpose,a.prepared_by,a.requested_approver,a.metric_path,a.rendered_value,a.destination)
else: decide(a.request,a.decision,a.actor)
Run from the project root. On Windows use py -3; elsewhere replace it with python3. These commands exercise both baseline paths:
py -3 steps/operate.py fixture --lane lab
py -3 steps/operate.py build --lane lab
py -3 steps/operate.py seal --lane lab --run-id LAB-BASE-001 --bootstrap-baseline
py -3 steps/operate.py request --run-id LAB-BASE-001 --request-id BREQ-LAB-001 --purpose DRIFT_BASELINE --prepared-by operator-a --requested-approver approver-b
py -3 steps/operate.py decide --request publications/requests/BREQ-LAB-001.json --decision APPROVED --actor approver-b
py -3 steps/operate.py fixture --lane company
py -3 steps/operate.py build --lane company
py -3 steps/operate.py seal --lane company --run-id CRM-BASE-001 --bootstrap-baseline
py -3 steps/operate.py request --run-id CRM-BASE-001 --request-id BREQ-CRM-001 --purpose DRIFT_BASELINE --prepared-by operator-a --requested-approver approver-b
py -3 steps/operate.py decide --request publications/requests/BREQ-CRM-001.json --decision APPROVED --actor approver-b
Then exercise both candidate and publication paths. Regenerating identical fixtures intentionally yields zero drift:
py -3 steps/operate.py fixture --lane lab
py -3 steps/operate.py build --lane lab
py -3 steps/operate.py seal --lane lab --run-id LAB-CAND-002 --baseline LAB-BASE-001
py -3 steps/operate.py request --run-id LAB-CAND-002 --request-id REQ-LAB-002 --purpose PUBLICATION --prepared-by operator-a --requested-approver approver-b --metric-path metrics.batch_mean_value_c.CEDAR-B --rendered-value "20.70 C" --destination "synthetic manuscript Figure 2 caption"
py -3 steps/operate.py decide --request publications/requests/REQ-LAB-002.json --decision APPROVED --actor approver-b
py -3 steps/operate.py fixture --lane company
py -3 steps/operate.py build --lane company
py -3 steps/operate.py seal --lane company --run-id CRM-CAND-002 --baseline CRM-BASE-001
py -3 steps/operate.py request --run-id CRM-CAND-002 --request-id REQ-CRM-002 --purpose PUBLICATION --prepared-by operator-a --requested-approver approver-b --metric-path metrics.qualified_activity_count --rendered-value "947" --destination "synthetic August board pack slide 4 cell B7"
py -3 steps/operate.py decide --request publications/requests/REQ-CRM-002.json --decision APPROVED --actor approver-b
Expected machine values are Lab count 4, mean 19.675, CEDAR-A = 18.65, and CEDAR-B = 20.7; Company count 1000, qualified count 947, with region and activity-type totals each reconciling to 947. operator_note and contact_email never enter safe rows. Inspect every manifest, report, request, and receipt rather than trusting terminal text.
This compact implementation records the executing script digest rather than requiring Git, so it runs exactly as copied. In production, additionally require a full committed revision, a clean tracked tree, reviewed configuration and policy files, encrypted approved storage, access control, object locking, independent deadline monitoring, backup/restore tests, and a source-native consistent snapshot. Those controls must not be claimed merely because this local drill passed.
7. What goes wrong
A mutable name replaces evidence
Symptom: an old report points to latest.json, whose bytes have changed. Fix: use unique source and run IDs, append-only storage, and receipts bound to exact digests. A mutable alias is convenience, never provenance.
“PASS” has no observations
Symptom: the manifest says green but has no schema fingerprint, observed values, segment totals, threshold, or baseline. Fix: retain check-results.json and drift-report.json; fail closed if either is absent or changed.
Nobody notices non-execution
Symptom: yesterday's result remains visible after today's scheduler fails. Fix: run an independent scheduler after the deadline. It must search for today's sealed manifest and overdue approval, create an owned alert, and rehearse acknowledgement and escalation. A pipeline cannot report that it never ran.
A hash is treated as truth or privacy
Symptom: a digest is offered as proof that a source was correct or anonymised. Fix: treat it only as byte identity. Keep source authority, quality, privacy, access, purpose, and retention as separate controls.
The operator approves a moving target
Symptom: one account prepares, approves, and publishes “latest.” Fix: enforce separate identities, bind the decision to request, manifest, output, metric, rendering, and destination, and deny changed bytes or self-approval.
Everything is retained forever
Symptom: snapshots, rejected runs, traces, replicas, and backups have no deletion date. Fix: classify each object, assign purpose and owner, account for legal holds and backup expiry, and preserve a non-sensitive deletion receipt after an independent verifier confirms deletion.
8. Trace one number yourself in 90 minutes
Choose Lab or Company. Do not publish the drill to a real manuscript, board pack, dashboard, or customer channel.
- 0–15 minutes: name metric, destination, source owner, operator, approver, deadline, baseline owner, retention owner, and stop conditions. Set thresholds before running.
- 15–35: run the baseline path, inspect safe fields and checks, approve with a separate identity, and verify baseline state points to unchanged receipt and manifest digests.
- 35–55: run the candidate path. Rename a schema field and prove build stops; restore it, change the Lab mean by over
2.0or Company qualified count by over15%, and prove sealing stops. - 55–70: restore the fixture, seal a passing candidate, request one exact destination, prove self-approval fails, then approve as the designated approver.
- 70–82: start from the displayed value. Recompute receipt, request, manifest, output, checks, drift, schema, script, parameter, and source identities. For Lab, confirm
(20.0 + 21.4) / 2 = 20.7. For Company, rerun the qualified count against the sealed synthetic snapshot. - 82–90: document retention, legal hold, replicas, backup expiry, deletion executor, and verifier. Have another operator repeat the trace without oral hints.
9. Exit check
Deliver exactly one artifact: one written provenance trace for one published number, from raw file to output.
It passes only when that single trace identifies the publication and displayed number; request and approval receipt; request, manifest, and output digests; machine value and calculation; contributing synthetic or approved records; source snapshot ID and digest; code and parameter digests; runtime; canonical schema fingerprint; per-check quality results; approved baseline and complete drift decision; expected-run monitor result or alert test; separate preparer and approver; retention, legal-hold, replica, backup, and verified-deletion obligations; and an independent operator's successful reconciliation.
Scripts, manifests, checks, reports, requests, receipts, and alerts are supporting evidence referenced inside that one written finding, not additional exit artifacts. A screenshot, latest link, placeholder digest, current-source query, or unreconciled value does not pass.
10. Rule to remember
You should be able to walk any number back to its file.
11. Further reading & tools
- Taught:
T10-L04- Pipelines over real data - supplies the read-only source, exact schema, quality gates, field boundary, and atomic working output extended here. - Taught:
T12-L05- Governance, evidence and handover - supplies ownership, continuing evidence, retention, and handover practices. - Taught: Python
sqlite3documentation (opens in a new tab) - primary reference for SQLite URI and query-only operation. - Taught: Python
hashlibdocumentation (opens in a new tab) - primary reference for SHA-256 evidence digests. - Taught: W3C PROV-O (opens in a new tab) - a standard model for richer entity, activity, agent, derivation, and attribution records.
- Catalogued: Git documentation (opens in a new tab) - add committed revision and clean-tree controls when adapting the drill.
- Catalogued: KNIME - a visual workflow option; provenance and publication authority remain operating contracts.
- Catalogued: Langfuse documentation (opens in a new tab) - tracing can add model-step evidence, but access, minimisation, retention, and deletion still need policy.
- Catalogued: Tools index - compare approved storage, orchestration, versioning, and observability tools after fixing the operating contract.