feat: auditable corpus manifest with tombstones and lifecycle guards (#25) #85
No files matched your search
@@ -33,6 +33,7 @@ jobs:
|
||||
run: |
|
||||
npm test
|
||||
python3 tests/staging-deploy.test.py -v
|
||||
python3 tests/corpus-manifest.test.py -v
|
||||
- name: Mobile browser acceptance
|
||||
run: |
|
||||
npm start > /tmp/timmy-server.log 2>&1 &
|
||||
|
||||
@@ -0,0 +1,61 @@
|
||||
# Corpus manifest contract
|
||||
|
||||
The training corpus is described by an **append-only** `manifest.jsonl`. One line is one event;
|
||||
nothing is ever rewritten in place, so the trail can be audited after a withdrawal. The
|
||||
implementation lives in [`scripts/corpus_manifest.py`](../scripts/corpus_manifest.py) and is
|
||||
driven by [`scripts/ingest_training_photo.py`](../scripts/ingest_training_photo.py).
|
||||
|
||||
## Sample record (`recordType: "sample"`, schema version 2)
|
||||
|
||||
| Field | Meaning |
|
||||
|---|---|
|
||||
| `contributionId` | Opaque id minted by the app's contribution receipt. The same id addresses every derived artifact and every later tombstone. |
|
||||
| `consent.version`, `consent.recordedAt` | The consent version the contributor accepted, and when. Duplicated as `consentVersion` for older readers. |
|
||||
| `source.kind`, `source.inputSha256` | How the bytes arrived (`cli-ingest`) and the digest of the caller's input. The caller's path is never recorded. |
|
||||
| `hashes.rawInput`, `hashes.derived` | Digest of the supplied bytes and of the re-encoded image that is actually stored. |
|
||||
| `derivation` | `format`, `maxDimension`, `metadataStripped`, and the `pipeline` version that produced the derived image. |
|
||||
| `review.status`, `review.reviewedAt` | `user-confirmed` or `clinician-reviewed`. A label set is never unlabelled-by-default. |
|
||||
| `split` | `train`, `validation`, or `test`, derived from the salted subject key so one contributor never straddles splits. |
|
||||
| `subjectKey` | 16-hex pseudonym derived from a salted subject id. Never a name, email, or account. |
|
||||
| `lifecycle.state`, `lifecycle.updatedAt` | `active` while the sample is usable for training. |
|
||||
| `imageId`, `relativePath` | Derived image digest and its dataset-relative path. |
|
||||
|
||||
## Tombstone record (`recordType: "tombstone"`)
|
||||
|
||||
A withdrawal never deletes history; it appends a tombstone that:
|
||||
|
||||
- keeps `contributionId` and `imageId` so the withdrawal can be matched to the original sample,
|
||||
- records `lifecycle.state: "revoked"`, the timestamp, and the reason,
|
||||
- lists `deletedArtifacts` — every derived copy it removed, addressed by contribution id with the
|
||||
digest that identifies the bytes,
|
||||
- deliberately drops `labels`, `subjectKey`, and `relativePath`: a deletion receipt keeps
|
||||
identifiers and hashes, not the health label or the person.
|
||||
|
||||
`revoke()` removes the derived image and any raw pre-derivation copy from the active dataset
|
||||
before appending the tombstone, and refuses any path that resolves outside the dataset root.
|
||||
|
||||
## Lifecycle rules (all covered by `tests/corpus-manifest.test.py`)
|
||||
|
||||
- `active` → `revoked` only, and only with a recorded reason.
|
||||
- `revoked` → `active` is refused: a revoked contribution cannot return to the active dataset.
|
||||
- Revoking an unknown contribution, or revoking twice, is refused and appends nothing.
|
||||
- A contribution id can be ingested once; the same derived image cannot carry two label sets.
|
||||
- `active_records()` returns only samples that are still `active` and were never tombstoned, so a
|
||||
trainer that reads through it cannot silently train on withdrawn data.
|
||||
|
||||
## Usage
|
||||
|
||||
```bash
|
||||
export TIMMY_DATASET_SALT=... # >= 8 chars, never committed
|
||||
python3 scripts/ingest_training_photo.py \
|
||||
--input photo.jpg --output-dir /srv/timmy-corpus \
|
||||
--subject-id subject-123 --consent-version research-v1 \
|
||||
--contribution-id "$CONTRIBUTION_ID" \
|
||||
--bristol-type 4 --color brown --quality good --reviewer user-confirmed
|
||||
|
||||
python3 scripts/ingest_training_photo.py --revoke "$CONTRIBUTION_ID" \
|
||||
--output-dir /srv/timmy-corpus --reason "subject withdrawal"
|
||||
```
|
||||
|
||||
`npm run test:corpus` runs the contract suite. Every fixture is synthetic; no real medical image
|
||||
is read, written, or committed.
|
||||
+3
-2
@@ -18,11 +18,12 @@
|
||||
"test:grounded-chat": "node tests/grounded-chat.acceptance.mjs",
|
||||
"test:photo-lifecycle": "node tests/photo-lifecycle.acceptance.mjs",
|
||||
"test:training-contribution": "node tests/training-contribution.acceptance.mjs",
|
||||
"check:syntax": "node --check app.js && node --check server.mjs && node --check service-worker.js && node --check src/analysis.js && node --check src/domain.js && node --check src/draft.js && node --check src/suggestion.js && node --check src/followup.js && node --check src/vision-readiness.js && node --check src/hermes-agent-service.js && node --check src/vision-config.js && node --check src/vision-service.js && node --check src/vision-corpus.js && node --check src/summary.js && node --check src/photo-lifecycle.js && node --check src/grounded-chat.js && node --check src/training-contribution.js && node --check scripts/record_release_demo.mjs && node --check tests/staging.acceptance.mjs && node --check tests/manual-fallback.acceptance.mjs && node --check tests/provisional-suggestion.acceptance.mjs && node --check tests/followup.acceptance.mjs && node --check tests/vision-refusal.acceptance.mjs && bash -n scripts/bootstrap_selfhost_smolvlm.sh && bash -n scripts/run_selfhost_smolvlm.sh && python3 -m py_compile scripts/ingest_training_photo.py scripts/build_release.py scripts/deploy_staging.py scripts/build_vision_corpus.py && node --check scripts/record_worker_benchmark.mjs && node --check tests/worker-readiness.acceptance.mjs && node --check tests/pattern-summary.acceptance.mjs && node --check tests/photo-lifecycle.acceptance.mjs && node --check tests/training-contribution.acceptance.mjs && node --check tests/portability.acceptance.mjs && node --check tests/grounded-chat.acceptance.mjs && bash -n scripts/worker_ready.sh && bash -n deploy/worker/entrypoint.sh && bash -n deploy/worker/build_image.sh && bash -n deploy/worker/run_worker_container.sh",
|
||||
"check:syntax": "node --check app.js && node --check server.mjs && node --check service-worker.js && node --check src/analysis.js && node --check src/domain.js && node --check src/draft.js && node --check src/suggestion.js && node --check src/followup.js && node --check src/vision-readiness.js && node --check src/hermes-agent-service.js && node --check src/vision-config.js && node --check src/vision-service.js && node --check src/vision-corpus.js && node --check src/summary.js && node --check src/photo-lifecycle.js && node --check src/grounded-chat.js && node --check src/training-contribution.js && node --check scripts/record_release_demo.mjs && node --check tests/staging.acceptance.mjs && node --check tests/manual-fallback.acceptance.mjs && node --check tests/provisional-suggestion.acceptance.mjs && node --check tests/followup.acceptance.mjs && node --check tests/vision-refusal.acceptance.mjs && bash -n scripts/bootstrap_selfhost_smolvlm.sh && bash -n scripts/run_selfhost_smolvlm.sh && python3 -m py_compile scripts/corpus_manifest.py scripts/ingest_training_photo.py scripts/build_release.py scripts/deploy_staging.py scripts/build_vision_corpus.py && node --check scripts/record_worker_benchmark.mjs && node --check tests/worker-readiness.acceptance.mjs && node --check tests/pattern-summary.acceptance.mjs && node --check tests/photo-lifecycle.acceptance.mjs && node --check tests/training-contribution.acceptance.mjs && node --check tests/portability.acceptance.mjs && node --check tests/grounded-chat.acceptance.mjs && bash -n scripts/worker_ready.sh && bash -n deploy/worker/entrypoint.sh && bash -n deploy/worker/build_image.sh && bash -n deploy/worker/run_worker_container.sh",
|
||||
"check:diff": "bash scripts/check_diff.sh",
|
||||
"start": "node server.mjs",
|
||||
"test:worker-readiness": "node tests/worker-readiness.acceptance.mjs",
|
||||
"test:portability": "node tests/portability.acceptance.mjs"
|
||||
"test:portability": "node tests/portability.acceptance.mjs",
|
||||
"test:corpus": "python3 tests/corpus-manifest.test.py"
|
||||
},
|
||||
"devDependencies": {
|
||||
"playwright": "^1.62.1"
|
||||
|
||||
@@ -0,0 +1,241 @@
|
||||
#!/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
|
||||
@@ -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
|
||||
|
||||
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)
|
||||
|
||||
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")
|
||||
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())
|
||||
@@ -0,0 +1,273 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Tests for the auditable corpus manifest: provenance, tombstones, and lifecycle rules.
|
||||
|
||||
Issue #25 promotes the ingest script into a corpus pipeline with an append-only manifest,
|
||||
revocation tombstones, and lifecycle transitions that cannot be walked backwards. Every
|
||||
fixture here is synthetic; no real photo bytes are read or written.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import importlib.util
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
import unittest
|
||||
|
||||
from PIL import Image
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
INGEST = ROOT / "scripts" / "ingest_training_photo.py"
|
||||
FIXTURE = ROOT / "tests" / "fixtures" / "synthetic-type4.jpg"
|
||||
CONTRIBUTION_ID = "6f1d0c9e-2c4a-4f3b-9d1e-8a7b6c5d4e3f"
|
||||
ISO_UTC = re.compile(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?\+00:00$")
|
||||
|
||||
|
||||
def load_structures():
|
||||
spec = importlib.util.spec_from_file_location("corpus_manifest", ROOT / "scripts" / "corpus_manifest.py")
|
||||
assert spec and spec.loader
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
structures = load_structures()
|
||||
|
||||
|
||||
def active_image_paths(out: Path) -> list[str]:
|
||||
return [record["relativePath"] for record in structures.active_records(structures.load_manifest(out / "manifest.jsonl"))]
|
||||
|
||||
|
||||
|
||||
def ingest(args: list[str], env: dict[str, str] | None = None) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(
|
||||
[sys.executable, str(INGEST), *args],
|
||||
cwd=ROOT,
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env={**os.environ, "TIMMY_DATASET_SALT": "test-only-salt", **(env or {})},
|
||||
)
|
||||
|
||||
|
||||
def ingest_default(out: Path, **overrides: str) -> subprocess.CompletedProcess:
|
||||
args = [
|
||||
"--input", str(FIXTURE),
|
||||
"--output-dir", str(out),
|
||||
"--subject-id", "subject-123",
|
||||
"--consent-version", "research-v1",
|
||||
"--contribution-id", CONTRIBUTION_ID,
|
||||
"--bristol-type", "4",
|
||||
"--color", "brown",
|
||||
"--quality", "good",
|
||||
"--reviewer", "user-confirmed",
|
||||
]
|
||||
for key, value in overrides.items():
|
||||
flag = "--" + key.replace("_", "-")
|
||||
if value is None:
|
||||
index = args.index(flag)
|
||||
del args[index:index + 2]
|
||||
continue
|
||||
args[args.index(flag) + 1] = value
|
||||
return ingest(args)
|
||||
|
||||
|
||||
def manifest_lines(out: Path) -> list[dict]:
|
||||
text = (out / "manifest.jsonl").read_text(encoding="utf-8")
|
||||
return [json.loads(line) for line in text.splitlines() if line.strip()]
|
||||
|
||||
|
||||
class IngestProvenanceTests(unittest.TestCase):
|
||||
def test_ingest_records_consent_source_hashes_derivation_review_split_and_lifecycle(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
run = ingest_default(out)
|
||||
self.assertEqual(run.returncode, 0, run.stderr)
|
||||
record = json.loads(run.stdout)
|
||||
raw_bytes = FIXTURE.read_bytes()
|
||||
|
||||
self.assertEqual(record["schemaVersion"], 2)
|
||||
self.assertEqual(record["contributionId"], CONTRIBUTION_ID)
|
||||
self.assertEqual(record["consentVersion"], "research-v1")
|
||||
self.assertEqual(record["consent"], {"version": "research-v1", "recordedAt": record["ingestedAt"]})
|
||||
self.assertRegex(record["consent"]["recordedAt"], ISO_UTC)
|
||||
|
||||
# Source is named, and the raw input only ever appears as a digest.
|
||||
self.assertEqual(record["source"], {
|
||||
"kind": "cli-ingest",
|
||||
"inputSha256": hashlib.sha256(raw_bytes).hexdigest(),
|
||||
})
|
||||
self.assertEqual(record["hashes"]["rawInput"], hashlib.sha256(raw_bytes).hexdigest())
|
||||
self.assertEqual(record["hashes"]["derived"], record["imageId"])
|
||||
|
||||
self.assertEqual(record["derivation"], {
|
||||
"format": "jpeg",
|
||||
"maxDimension": 1200,
|
||||
"metadataStripped": True,
|
||||
"pipeline": "timmy-ingest/2",
|
||||
})
|
||||
self.assertEqual(record["review"], {"status": "user-confirmed", "reviewedAt": record["ingestedAt"]})
|
||||
self.assertIn(record["split"], ("train", "validation", "test"))
|
||||
self.assertEqual(record["lifecycle"], {"state": "active", "updatedAt": record["ingestedAt"]})
|
||||
self.assertEqual(record["recordType"], "sample")
|
||||
|
||||
# Provenance never leaks the caller's path or the source bytes.
|
||||
self.assertNotIn("inputPath", record)
|
||||
on_disk = (out / "manifest.jsonl").read_text(encoding="utf-8")
|
||||
self.assertNotIn(str(FIXTURE), on_disk)
|
||||
self.assertNotIn(str(out), on_disk)
|
||||
self.assertEqual(manifest_lines(out), [record])
|
||||
self.assertTrue((out / record["relativePath"]).is_file())
|
||||
|
||||
|
||||
class RevocationTests(unittest.TestCase):
|
||||
def test_revocation_appends_a_tombstone_and_removes_the_bytes_from_active_datasets(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
self.assertEqual(ingest_default(out).returncode, 0)
|
||||
sample = manifest_lines(out)[0]
|
||||
derived = out / sample["relativePath"]
|
||||
self.assertTrue(derived.is_file())
|
||||
# Some pipelines keep the pre-derivation copy; revocation must clear that too.
|
||||
raw_dir = out / "raw"
|
||||
raw_dir.mkdir()
|
||||
raw_copy = raw_dir / f"{sample['hashes']['rawInput']}.jpg"
|
||||
raw_copy.write_bytes(FIXTURE.read_bytes())
|
||||
|
||||
run = ingest(["--revoke", CONTRIBUTION_ID, "--output-dir", str(out), "--reason", "subject withdrawal"])
|
||||
self.assertEqual(run.returncode, 0, run.stderr)
|
||||
tombstone = json.loads(run.stdout)
|
||||
|
||||
self.assertEqual(tombstone["schemaVersion"], 2)
|
||||
self.assertEqual(tombstone["recordType"], "tombstone")
|
||||
self.assertEqual(tombstone["contributionId"], CONTRIBUTION_ID)
|
||||
self.assertEqual(tombstone["imageId"], sample["imageId"])
|
||||
self.assertEqual(tombstone["lifecycle"]["state"], "revoked")
|
||||
self.assertRegex(tombstone["lifecycle"]["updatedAt"], ISO_UTC)
|
||||
self.assertEqual(tombstone["lifecycle"]["reason"], "subject withdrawal")
|
||||
|
||||
# The tombstone addresses every derived artifact by contribution id, and says what went.
|
||||
self.assertEqual(tombstone["deletedArtifacts"], [
|
||||
{"relativePath": sample["relativePath"], "sha256": sample["hashes"]["derived"], "deleted": True},
|
||||
{"relativePath": f"raw/{sample['hashes']['rawInput']}.jpg", "sha256": sample["hashes"]["rawInput"], "deleted": True},
|
||||
])
|
||||
self.assertEqual(tombstone["contributionId"], CONTRIBUTION_ID)
|
||||
|
||||
# No byte-bearing copy survives anywhere in the dataset tree.
|
||||
self.assertFalse(derived.exists())
|
||||
self.assertFalse(raw_copy.exists())
|
||||
survivors = []
|
||||
for path in out.rglob("*"):
|
||||
if path.is_file() and path.name != "manifest.jsonl":
|
||||
survivors.append((str(path.relative_to(out)), hashlib.sha256(path.read_bytes()).hexdigest()))
|
||||
self.assertEqual(survivors, [])
|
||||
|
||||
# History is append-only: the sample record is untouched, the tombstone follows it.
|
||||
history = manifest_lines(out)
|
||||
self.assertEqual(len(history), 2)
|
||||
self.assertEqual(history[0], sample)
|
||||
self.assertEqual(history[1], tombstone)
|
||||
|
||||
# A deletion receipt keeps identifiers and hashes, never the label or subject key.
|
||||
self.assertNotIn("labels", tombstone)
|
||||
self.assertNotIn("subjectKey", tombstone)
|
||||
self.assertNotIn("relativePath", tombstone)
|
||||
|
||||
|
||||
class ActiveRecordTests(unittest.TestCase):
|
||||
def test_only_unrevoked_samples_are_active(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
self.assertEqual(ingest_default(out).returncode, 0)
|
||||
self.assertEqual(active_image_paths(out), [f"images/{manifest_lines(out)[0]['imageId']}.jpg"])
|
||||
self.assertEqual(ingest(["--revoke", CONTRIBUTION_ID, "--output-dir", str(out), "--reason", "gone"]).returncode, 0)
|
||||
self.assertEqual(active_image_paths(out), [])
|
||||
|
||||
|
||||
class IllegalTransitionTests(unittest.TestCase):
|
||||
def test_a_second_revocation_is_refused_and_appends_nothing(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
self.assertEqual(ingest_default(out).returncode, 0)
|
||||
self.assertEqual(ingest(["--revoke", CONTRIBUTION_ID, "--output-dir", str(out), "--reason", "first"]).returncode, 0)
|
||||
history = manifest_lines(out)
|
||||
self.assertEqual(len(history), 2)
|
||||
|
||||
again = ingest(["--revoke", CONTRIBUTION_ID, "--output-dir", str(out), "--reason", "again"])
|
||||
self.assertNotEqual(again.returncode, 0)
|
||||
self.assertRegex(again.stderr, "already revoked")
|
||||
self.assertEqual(manifest_lines(out), history)
|
||||
|
||||
def test_revoking_an_unknown_contribution_is_refused(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
self.assertEqual(ingest_default(out).returncode, 0)
|
||||
history = manifest_lines(out)
|
||||
|
||||
unknown = ingest(["--revoke", "11111111-2222-3333-4444-555555555555", "--output-dir", str(out), "--reason", "nope"])
|
||||
self.assertNotEqual(unknown.returncode, 0)
|
||||
self.assertRegex(unknown.stderr, "no ingested sample carries contribution")
|
||||
self.assertEqual(manifest_lines(out), history)
|
||||
self.assertTrue((out / history[0]["relativePath"]).is_file())
|
||||
|
||||
def test_a_contribution_cannot_be_ingested_twice(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
self.assertEqual(ingest_default(out).returncode, 0)
|
||||
history = manifest_lines(out)
|
||||
|
||||
duplicate = ingest_default(out, color="green")
|
||||
self.assertNotEqual(duplicate.returncode, 0)
|
||||
self.assertRegex(duplicate.stderr, "already ingested")
|
||||
self.assertEqual(manifest_lines(out), history)
|
||||
|
||||
def test_a_refused_duplicate_leaves_no_orphan_image_behind(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
self.assertEqual(ingest_default(out).returncode, 0)
|
||||
history = manifest_lines(out)
|
||||
|
||||
# Same contribution id, different pixels: the derived digest differs, so a guard that
|
||||
# runs after the write would strand a second image in the active dataset.
|
||||
variant = out.parent / "variant-type4.jpg"
|
||||
with Image.open(FIXTURE) as source:
|
||||
changed = source.convert("RGB")
|
||||
changed.putpixel((4, 4), (255, 0, 0))
|
||||
changed.save(variant, "JPEG", quality=95)
|
||||
duplicate = ingest_default(out, input=str(variant))
|
||||
self.assertNotEqual(duplicate.returncode, 0)
|
||||
self.assertRegex(duplicate.stderr, "already ingested")
|
||||
|
||||
self.assertEqual(manifest_lines(out), history)
|
||||
images = sorted(path.name for path in (out / "images").iterdir())
|
||||
self.assertEqual(images, [f"{history[0]['imageId']}.jpg"])
|
||||
|
||||
|
||||
def test_lifecycle_transitions_cannot_walk_backwards(self):
|
||||
with tempfile.TemporaryDirectory() as raw:
|
||||
out = Path(raw)
|
||||
self.assertEqual(ingest_default(out).returncode, 0)
|
||||
sample = manifest_lines(out)[0]
|
||||
at = sample["ingestedAt"]
|
||||
|
||||
revoked = structures.transition(sample, "revoked", at, "withdrawal")
|
||||
self.assertEqual(revoked["lifecycle"]["state"], "revoked")
|
||||
self.assertEqual(revoked["lifecycle"]["reason"], "withdrawal")
|
||||
self.assertEqual(sample["lifecycle"]["state"], "active", "transition must not mutate its input")
|
||||
|
||||
for args, expected in (
|
||||
((revoked, "active", at), "cannot return to the active dataset"),
|
||||
((sample, "revoked", at, " "), "recorded reason"),
|
||||
((sample, "archived", at), "unknown lifecycle state"),
|
||||
(({**sample, "lifecycle": {"state": "active", "updatedAt": at}}, "active", at), "already active"),
|
||||
):
|
||||
with self.assertRaises(structures.IllegalTransition) as raised:
|
||||
structures.transition(*args)
|
||||
self.assertRegex(str(raised.exception), expected)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
@@ -14,6 +14,7 @@ test('training ingest strips metadata, derives a pseudonymous split, and writes
|
||||
'--output-dir', out,
|
||||
'--subject-id', 'subject-123',
|
||||
'--consent-version', 'research-v1',
|
||||
'--contribution-id', '6f1d0c9e-2c4a-4f3b-9d1e-8a7b6c5d4e3f',
|
||||
'--bristol-type', '4',
|
||||
'--color', 'brown',
|
||||
'--quality', 'good',
|
||||
@@ -45,10 +46,11 @@ test('training ingest rejects records without explicit consent or human review',
|
||||
'--input', 'tests/fixtures/synthetic-type4.jpg',
|
||||
'--output-dir', '/tmp/should-not-exist-timmy',
|
||||
'--subject-id', 'subject-123',
|
||||
'--contribution-id', '6f1d0c9e-2c4a-4f3b-9d1e-8a7b6c5d4e3f',
|
||||
'--bristol-type', '4',
|
||||
'--color', 'brown',
|
||||
'--quality', 'good',
|
||||
], { encoding: 'utf8', env: { ...process.env, TIMMY_DATASET_SALT: 'test-only-salt' } });
|
||||
assert.notEqual(run.status, 0);
|
||||
assert.match(run.stderr, /consent|reviewer/i);
|
||||
assert.match(run.stderr, /consent/i);
|
||||
});
|
||||
Reference in new issue
Block a user