feat(corpus): auditable manifest provenance, tombstones, and lifecycle guards (#25)

Promote the ingest script into an auditable corpus pipeline. Each sample now
records consent (version + timestamp), source kind and raw-input digest, both
hashes, the derivation pipeline, review status, split, and lifecycle state, with
the caller's path never written down.

Withdrawal is a new `--revoke <contribution-id> --reason` operation: it deletes
the derived image and any raw pre-derivation copy, then appends a tombstone that
addresses each removed artifact by contribution id and digest while dropping
labels, subject key, and paths. History is append-only, so the original sample
record survives the withdrawal it documents.

Illegal moves are refused instead of silently applied: revoked -> active,
revoking an unknown or already-revoked contribution, an unknown state, a
duplicate contribution id, and a second label set for the same derived image.
The duplicate guard runs before any byte is written, so a refused ingest cannot
strand an image in the active dataset.

Tests written first and watched fail (tests/corpus-manifest.test.py): provenance,
tombstone + byte removal, active-record filtering, and the transition guards.
The transition helper itself passed on first write and is reported as a
hold-out check rather than a RED cycle. Wired into the quality workflow and
check:syntax; contract documented in docs/CORPUS-MANIFEST.md.

No browser surface changed, so record_release_demo.mjs is untouched and claims
nothing new: the nearest honest visible effect is the withdrawn-receipt
tombstone already recorded for #24.

scripts/corpus_manifest.py

@@ -0,0 +1,242 @@
+#!/usr/bin/env python3
+"""Immutable, auditable corpus manifest for Timmy training contributions.
+
+Every ingested sample is described by one append-only manifest record: who consented, under
+which consent version, where the bytes came from, the hashes of the raw input and the derived
+image, how the derived image was produced, who reviewed the labels, which split it belongs to,
+and where it sits in its lifecycle. Nothing in this module rewrites history: records are only
+ever appended, and a withdrawal is a tombstone, not a deletion of the trail.
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+from pathlib import Path
+
+SCHEMA_VERSION = 2
+PIPELINE = "timmy-ingest/2"
+
+SAMPLE = "sample"
+TOMBSTONE = "tombstone"
+RECORD_TYPES = frozenset({SAMPLE, TOMBSTONE})
+
+ACTIVE = "active"
+REVOKED = "revoked"
+STATES = frozenset({ACTIVE, REVOKED})
+
+SPLITS = ("train", "validation", "test")
+REVIEWERS = ("user-confirmed", "clinician-reviewed")
+COLORS = ("brown", "green", "yellow", "pale", "red", "black")
+QUALITIES = ("good", "fair", "poor")
+
+_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+_CONTRIBUTION_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$")
+
+
+class CorpusError(Exception):
+    """A record cannot be built or written as asked."""
+
+
+class IllegalTransition(CorpusError):
+    """A lifecycle change that must never happen is being attempted."""
+
+
+def sha256_hex(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def _require_sha256(value: object, field: str) -> str:
+    if not isinstance(value, str) or not _SHA256.match(value):
+        raise CorpusError(f"{field} must be a lowercase hex sha256 digest")
+    return value
+
+
+def _require_iso(value: object, field: str) -> str:
+    if not isinstance(value, str) or not re.match(r"^\d{4}-\d{2}-\d{2}T", value):
+        raise CorpusError(f"{field} must be an ISO-8601 timestamp")
+    return value
+
+
+def build_record(
+    *,
+    contribution_id: str,
+    consent_version: str,
+    subject_key: str,
+    split: str,
+    labels: dict,
+    review_status: str,
+    derived_sha256: str,
+    raw_sha256: str,
+    ingested_at: str,
+    source_kind: str = "cli-ingest",
+) -> dict:
+    """Build one active sample record carrying its full provenance."""
+    if not isinstance(contribution_id, str) or not _CONTRIBUTION_ID.match(contribution_id):
+        raise CorpusError("contribution id is required and must be a stable opaque identifier")
+    if not isinstance(consent_version, str) or not consent_version.strip():
+        raise CorpusError("consent version is required")
+    if not isinstance(subject_key, str) or not re.match(r"^[0-9a-f]{16}$", subject_key):
+        raise CorpusError("subject key must be the 16-character pseudonymous digest")
+    if split not in SPLITS:
+        raise CorpusError(f"split must be one of {', '.join(SPLITS)}")
+    if review_status not in REVIEWERS:
+        raise CorpusError(f"review status must be one of {', '.join(REVIEWERS)}")
+    if not isinstance(source_kind, str) or not source_kind.strip():
+        raise CorpusError("source kind is required")
+    if not isinstance(labels, dict) or set(labels) != {"bristolType", "color", "quality"}:
+        raise CorpusError("labels must carry bristolType, color, and quality")
+    if labels["bristolType"] not in range(1, 8):
+        raise CorpusError("bristolType must be 1-7")
+    if labels["color"] not in COLORS or labels["quality"] not in QUALITIES:
+        raise CorpusError("color and quality must be from the documented vocabularies")
+    derived = _require_sha256(derived_sha256, "derived sha256")
+    raw = _require_sha256(raw_sha256, "raw input sha256")
+    stamped = _require_iso(ingested_at, "ingestedAt")
+
+    return {
+        "schemaVersion": SCHEMA_VERSION,
+        "recordType": SAMPLE,
+        "contributionId": contribution_id,
+        "imageId": derived,
+        "relativePath": f"images/{derived}.jpg",
+        "subjectKey": subject_key,
+        "split": split,
+        "consentVersion": consent_version.strip(),
+        "consent": {"version": consent_version.strip(), "recordedAt": stamped},
+        "source": {"kind": source_kind, "inputSha256": raw},
+        "hashes": {"rawInput": raw, "derived": derived},
+        "derivation": {
+            "format": "jpeg",
+            "maxDimension": 1200,
+            "metadataStripped": True,
+            "pipeline": PIPELINE,
+        },
+        "labels": {
+            "bristolType": labels["bristolType"],
+            "color": labels["color"],
+            "quality": labels["quality"],
+        },
+        "review": {"status": review_status, "reviewedAt": stamped},
+        "lifecycle": {"state": ACTIVE, "updatedAt": stamped},
+        "ingestedAt": stamped,
+    }
+
+
+def assert_admissible(manifest_path: Path, record: dict) -> None:
+    """Refuse a sample that duplicates a contribution or re-labels the same derived image."""
+    if record.get("recordType", SAMPLE) != SAMPLE:
+        return
+    for prior in load_manifest(manifest_path):
+        if prior.get("recordType", SAMPLE) != SAMPLE:
+            continue
+        if prior.get("contributionId") == record.get("contributionId"):
+            raise IllegalTransition(f"contribution {record['contributionId']} was already ingested")
+        if prior.get("imageId") == record.get("imageId") and prior.get("labels") != record.get("labels"):
+            raise CorpusError("the same derived image cannot carry two different label sets")
+
+
+def append_record(manifest_path: Path, record: dict) -> None:
+    """Append one record to the manifest without touching what is already there."""
+    assert_admissible(manifest_path, record)
+    manifest_path.parent.mkdir(parents=True, exist_ok=True)
+    with manifest_path.open("a", encoding="utf-8") as manifest:
+        manifest.write(json.dumps(record, separators=(",", ":"), sort_keys=False) + "\n")
+
+
+def load_manifest(manifest_path: Path) -> list[dict]:
+    """Read the append-only manifest in order. A missing manifest is an empty corpus."""
+    if not manifest_path.exists():
+        return []
+    with manifest_path.open("r", encoding="utf-8") as manifest:
+        return [json.loads(line) for line in manifest.read().splitlines() if line.strip()]
+
+
+def transition(record: dict, state: str, at: str, reason: str | None = None) -> dict:
+    """Return a copy of the record in the next lifecycle state, or refuse the move."""
+    current = record.get("lifecycle", {}).get("state")
+    if state not in STATES:
+        raise IllegalTransition(f"unknown lifecycle state: {state!r}")
+    if current not in STATES:
+        raise IllegalTransition(f"record carries no legal lifecycle state: {current!r}")
+    if current == state:
+        raise IllegalTransition(f"record is already {state}")
+    if current == REVOKED:
+        raise IllegalTransition("a revoked contribution cannot return to the active dataset")
+    _require_iso(at, "lifecycle timestamp")
+    if state == REVOKED and (not isinstance(reason, str) or not reason.strip()):
+        raise IllegalTransition("revocation needs a recorded reason")
+    updated = {**record, "lifecycle": {"state": state, "updatedAt": at}}
+    if state == REVOKED:
+        updated["lifecycle"]["reason"] = reason.strip()
+    return updated
+
+
+def active_records(records: list[dict]) -> list[dict]:
+    """Samples still usable for training: active, and never revoked afterwards."""
+    revoked = {
+        record.get("contributionId")
+        for record in records
+        if record.get("recordType") == TOMBSTONE or record.get("lifecycle", {}).get("state") == REVOKED
+    }
+    return [
+        record
+        for record in records
+        if record.get("recordType", SAMPLE) == SAMPLE
+        and record.get("lifecycle", {}).get("state") == ACTIVE
+        and record.get("contributionId") not in revoked
+    ]
+
+
+def _safe_dataset_path(dataset_root: Path, relative: str) -> Path:
+    path = (dataset_root / relative).resolve()
+    if not path.is_relative_to(dataset_root.resolve()):
+        raise CorpusError(f"refusing to touch a path outside the dataset: {relative}")
+    return path
+
+
+def revoke(dataset_root: Path, contribution_id: str, at: str, reason: str) -> dict:
+    """Revoke one contribution: append a tombstone and delete its bytes from the dataset."""
+    manifest_path = dataset_root / "manifest.jsonl"
+    records = load_manifest(manifest_path)
+    samples = [
+        record
+        for record in records
+        if record.get("recordType", SAMPLE) == SAMPLE and record.get("contributionId") == contribution_id
+    ]
+    if not isinstance(reason, str) or not reason.strip():
+        raise IllegalTransition("revocation needs a recorded reason")
+    if any(
+        record.get("recordType") == TOMBSTONE and record.get("contributionId") == contribution_id
+        for record in records
+    ):
+        raise IllegalTransition(f"contribution {contribution_id} is already revoked")
+    if not samples:
+        raise IllegalTransition(f"no ingested sample carries contribution {contribution_id}")
+    sample = samples[-1]
+    if sample.get("lifecycle", {}).get("state") != ACTIVE:
+        raise IllegalTransition(f"contribution {contribution_id} is not active")
+
+    deleted = []
+    candidates = [(sample["relativePath"], sample["hashes"]["derived"])]
+    # A raw pre-derivation copy is identified by the raw input digest, never by a user path.
+    for path in sorted((dataset_root / "raw").glob(f"{sample['hashes']['rawInput']}*")):
+        candidates.append((str(path.relative_to(dataset_root)), sample["hashes"]["rawInput"]))
+    for relative, digest in candidates:
+        target = _safe_dataset_path(dataset_root, relative)
+        existed = target.is_file()
+        if existed:
+            target.unlink()
+        deleted.append({"relativePath": relative, "sha256": digest, "deleted": existed})
+
+    tombstone = {
+        "schemaVersion": SCHEMA_VERSION,
+        "recordType": TOMBSTONE,
+        "contributionId": contribution_id,
+        "imageId": sample["imageId"],
+        "deletedArtifacts": deleted,
+        "lifecycle": {"state": REVOKED, "updatedAt": _require_iso(at, "lifecycle timestamp"), "reason": reason.strip()},
+    }
+    append_record(manifest_path, tombstone)
+    return tombstone
+

scripts/ingest_training_photo.py

@@ -12,6 +12,8 @@ from pathlib import Path
 
 from PIL import Image, ImageOps
 
+from corpus_manifest import CorpusError, append_record, assert_admissible, build_record, revoke, sha256_hex
+
 COLORS = ("brown", "green", "yellow", "pale", "red", "black")
 QUALITIES = ("good", "fair", "poor")
 REVIEWERS = ("user-confirmed", "clinician-reviewed")
@@ -19,27 +21,62 @@ REVIEWERS = ("user-confirmed", "clinician-reviewed")
 
 def parser() -> argparse.ArgumentParser:
     p = argparse.ArgumentParser()
-    p.add_argument("--input", required=True)
+    p.add_argument("--input")
     p.add_argument("--output-dir", required=True)
-    p.add_argument("--subject-id", required=True)
-    p.add_argument("--consent-version", required=True)
-    p.add_argument("--bristol-type", required=True, type=int, choices=range(1, 8))
-    p.add_argument("--color", required=True, choices=COLORS)
-    p.add_argument("--quality", required=True, choices=QUALITIES)
-    p.add_argument("--reviewer", required=True, choices=REVIEWERS)
+    p.add_argument("--subject-id")
+    p.add_argument("--consent-version")
+    p.add_argument("--contribution-id")
+    p.add_argument("--bristol-type", type=int, choices=range(1, 8))
+    p.add_argument("--color", choices=COLORS)
+    p.add_argument("--quality", choices=QUALITIES)
+    p.add_argument("--reviewer", choices=REVIEWERS)
+    # Revocation is a corpus operation, not an ingestion: it needs the dataset and a reason.
+    p.add_argument("--revoke", metavar="CONTRIBUTION_ID")
+    p.add_argument("--reason")
     return p
 
 
+def require_ingest_arguments(args: argparse.Namespace) -> None:
+    if not args.consent_version or not args.consent_version.strip():
+        raise SystemExit("consent version is required")
+    if not args.reviewer:
+        raise SystemExit("a human review status is required")
+    missing = [
+        flag
+        for flag, value in (
+            ("--input", args.input),
+            ("--subject-id", args.subject_id),
+            ("--contribution-id", args.contribution_id),
+            ("--bristol-type", args.bristol_type),
+            ("--color", args.color),
+            ("--quality", args.quality),
+        )
+        if value is None or value == ""
+    ]
+    if missing:
+        raise SystemExit(f"ingesting a sample also needs {', '.join(missing)}")
+
+
 def main() -> int:
     args = parser().parse_args()
+    if args.revoke:
+        if not args.reason:
+            raise SystemExit("revoking a contribution needs a recorded reason")
+        try:
+            tombstone = revoke(Path(args.output_dir), args.revoke, datetime.now(timezone.utc).isoformat(), args.reason)
+        except CorpusError as exc:
+            raise SystemExit(str(exc)) from exc
+        print(json.dumps(tombstone))
+        return 0
+
+    require_ingest_arguments(args)
     salt = os.environ.get("TIMMY_DATASET_SALT", "")
     if len(salt) < 8:
         raise SystemExit("TIMMY_DATASET_SALT must contain at least 8 characters")
-    if not args.consent_version.strip():
-        raise SystemExit("consent version is required")
 
     try:
-        with Image.open(args.input) as source:
+        raw_bytes = Path(args.input).read_bytes()
+        with Image.open(io.BytesIO(raw_bytes)) as source:
             image = ImageOps.exif_transpose(source).convert("RGB")
             image.thumbnail((1200, 1200))
             buffer = io.BytesIO()
@@ -55,37 +92,34 @@ def main() -> int:
 
     output = Path(args.output_dir)
     images = output / "images"
-    images.mkdir(parents=True, exist_ok=True)
     relative_path = f"images/{image_id}.jpg"
     target = output / relative_path
-    if not target.exists():
-        target.write_bytes(data)
 
-    record = {
-        "schemaVersion": 1,
-        "imageId": image_id,
-        "relativePath": relative_path,
-        "subjectKey": subject_key,
-        "split": split,
-        "consentVersion": args.consent_version.strip(),
-        "ingestedAt": datetime.now(timezone.utc).isoformat(),
-        "labels": {
-            "bristolType": args.bristol_type,
-            "color": args.color,
-            "quality": args.quality,
-        },
-        "review": {"status": args.reviewer},
-        "derivation": {
-            "format": "jpeg",
-            "maxDimension": 1200,
-            "metadataStripped": True,
-        },
-    }
-    with (output / "manifest.jsonl").open("a", encoding="utf-8") as manifest:
-        manifest.write(json.dumps(record, separators=(",", ":")) + "\n")
+    now = datetime.now(timezone.utc).isoformat()
+    record = build_record(
+        contribution_id=args.contribution_id,
+        consent_version=args.consent_version,
+        subject_key=subject_key,
+        split=split,
+        labels={"bristolType": args.bristol_type, "color": args.color, "quality": args.quality},
+        review_status=args.reviewer,
+        derived_sha256=image_id,
+        raw_sha256=sha256_hex(raw_bytes),
+        ingested_at=now,
+    )
+    try:
+        # Refuse a duplicate contribution before any byte lands in the active dataset.
+        assert_admissible(output / "manifest.jsonl", record)
+        images.mkdir(parents=True, exist_ok=True)
+        if not target.exists():
+            target.write_bytes(data)
+        append_record(output / "manifest.jsonl", record)
+    except CorpusError as exc:
+        raise SystemExit(str(exc)) from exc
     print(json.dumps(record))
     return 0
 
 
+
 if __name__ == "__main__":
     sys.exit(main())