diff --git a/README.md b/README.md index 4ead8e7..71eb649 100644 --- a/README.md +++ b/README.md @@ -1008,3 +1008,7 @@ Versioned container images are available from the [NewsLens GitHub Container Reg NewsLens source code is released under the [MIT License](LICENSE). The Microsoft MIND dataset is governed by separate Microsoft Research License Terms and is not redistributed by this repository. + +## Implementation update + +See [implementation and evidence limits](docs/model-lifecycle.md). diff --git a/docs/model-lifecycle.md b/docs/model-lifecycle.md new file mode 100644 index 0000000..4d1e445 --- /dev/null +++ b/docs/model-lifecycle.md @@ -0,0 +1,41 @@ +# Local model lifecycle + +This POSIX runner connects immutable MIND snapshots to chronological recipe validation, +content-addressed receipts, artifact integrity checks, manual promotion and rollback. +Use a training/development snapshot that excludes the official final holdout. +The cutoff filters behavior events; the caller must supply the catalog available at +that cutoff. Article publication times cannot be reconstructed from these TSV files. + +```sh +python scripts/manage_lifecycle.py --root runs/lifecycle train --snapshot data/snapshot --cutoff 2019-11-14 --minimum-ndcg 0.3 --minimum-validation 100 +python scripts/manage_lifecycle.py --root runs/lifecycle promote RUN_ID +NEWSLENS_RELEASE_ROOT=runs/lifecycle uvicorn newslens.api.app:app +python scripts/manage_lifecycle.py --root runs/lifecycle rollback +``` + +Replace RUN_ID with the receipt ID. Repeating the same specification reuses its +verified result. Use successive immutable snapshots/cutoffs for backfills; each +produces a separate receipt. Rejected candidates cannot be promoted. Release +pointers are written atomically under a process lock. Restart API workers after +promotion or rollback; each worker resolves and verifies the pointer at startup. +Set only one of NEWSLENS_RELEASE_ROOT and NEWSLENS_ARTIFACT_PATH. + +The gate assesses the training recipe on a chronological validation split and +then refits on available snapshot events. It does not establish final-test quality +of that refit. This local implementation does not establish cloud orchestration, +a long-running production deployment, user lift, or a sustained operational SLO. +Keep the root writable only by trusted operators; artifact loading uses joblib. + +## Repeatable isolated release drill + +```sh +python scripts/rehearse_lifecycle.py --snapshot data/MINDsmall_train --root runs/drill-01 --cutoffs 2019-11-13 2019-11-14 --minimum-ndcg 0.3 --minimum-validation 100 +``` + +Choose cutoffs supported by the snapshot before evaluating. The script requires a +new registry, trains two candidates, verifies retry idempotency, promotes each and +rolls back to the first. `rehearsal.json` records measured step durations. The +integration test exercises the complete transition using explicitly artificial +fixtures. The drill does not start an HTTP server or prove service recovery/SLOs. +The documented Microsoft training download returned HTTP 409 (public access +prohibited) during this follow-up; a permitted local snapshot is still needed. diff --git a/scripts/manage_lifecycle.py b/scripts/manage_lifecycle.py new file mode 100755 index 0000000..42440c7 --- /dev/null +++ b/scripts/manage_lifecycle.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +"""Train an immutable snapshot, promote/rollback, or resolve a serving artifact.""" + +import argparse +import json +from pathlib import Path + +from newslens.operations.lifecycle import release, serving_path, train + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--root", type=Path, required=True) + sub = p.add_subparsers(dest="command", required=True) + t = sub.add_parser("train") + t.add_argument("--snapshot", type=Path, required=True) + t.add_argument("--cutoff", required=True) + t.add_argument("--minimum-ndcg", type=float, required=True) + t.add_argument("--minimum-validation", type=int, default=100) + promote = sub.add_parser("promote") + promote.add_argument("run_id") + sub.add_parser("rollback") + sub.add_parser("serving-path") + a = p.parse_args() + if a.command == "train": + result = train( + a.snapshot, + a.root, + cutoff=a.cutoff, + minimum_ndcg=a.minimum_ndcg, + minimum_validation=a.minimum_validation, + ) + elif a.command == "serving-path": + print(serving_path(a.root)) + return + else: + result = release(a.root, getattr(a, "run_id", None), rollback=a.command == "rollback") + print(json.dumps(result, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/rehearse_lifecycle.py b/scripts/rehearse_lifecycle.py new file mode 100644 index 0000000..0a234fe --- /dev/null +++ b/scripts/rehearse_lifecycle.py @@ -0,0 +1,85 @@ +"""Rehearse verified release transitions in an isolated local registry.""" + +import argparse +import json +import time +from pathlib import Path + +from newslens.operations.lifecycle import release, serving_path, train + + +def rehearse(snapshot, root, cutoffs, minimum_ndcg, minimum_validation): + if root.exists(): + raise ValueError("use a new isolated registry; never rehearse on a live release root") + if len(cutoffs) != 2 or cutoffs[0] == cutoffs[1]: + raise ValueError("two distinct snapshot cutoffs required") + events = [] + candidates = [] + for cutoff in cutoffs: + started = time.perf_counter() + args = { + "cutoff": cutoff, + "minimum_ndcg": minimum_ndcg, + "minimum_validation": minimum_validation, + } + receipt = train(snapshot, root, **args) + if receipt["status"] != "candidate": + raise ValueError("candidate rejected; do not lower gates after inspecting results") + assert train(snapshot, root, **args) == receipt + candidates.append(receipt["run_id"]) + events.append( + { + "step": "train_and_verified_retry", + "run_id": receipt["run_id"], + "elapsed_seconds": time.perf_counter() - started, + } + ) + for candidate in candidates: + started = time.perf_counter() + release(root, candidate) + path = serving_path(root) + events.append( + { + "step": "promote_and_resolve", + "run_id": candidate, + "artifact_path": str(path), + "elapsed_seconds": time.perf_counter() - started, + } + ) + started = time.perf_counter() + state = release(root, rollback=True) + assert state["current"] == candidates[0] + serving_path(root) + events.append( + { + "step": "rollback_and_resolve", + "run_id": state["current"], + "elapsed_seconds": time.perf_counter() - started, + } + ) + report = { + "status": "completed", + "scope": "local registry drill; no HTTP uptime or user lift claim", + "events": events, + } + (root / "rehearsal.json").write_text(json.dumps(report, indent=2) + "\n") + return report + + +def main(): + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("--snapshot", type=Path, required=True) + p.add_argument("--root", type=Path, required=True) + p.add_argument("--cutoffs", nargs=2, required=True) + p.add_argument("--minimum-ndcg", type=float, required=True) + p.add_argument("--minimum-validation", type=int, default=100) + a = p.parse_args() + print( + json.dumps( + rehearse(a.snapshot, a.root, a.cutoffs, a.minimum_ndcg, a.minimum_validation), indent=2 + ) + ) + + +if __name__ == "__main__": + main() diff --git a/src/newslens/api/settings.py b/src/newslens/api/settings.py index 537ba1b..ea12346 100644 --- a/src/newslens/api/settings.py +++ b/src/newslens/api/settings.py @@ -29,6 +29,16 @@ def from_environment(cls) -> ApiSettings: raw_artifact_path = os.getenv(ARTIFACT_PATH_ENVIRONMENT_VARIABLE) + release_root = os.getenv("NEWSLENS_RELEASE_ROOT") + if release_root is not None: + if not release_root.strip() or raw_artifact_path is not None: + raise ApiSettingsError( + "Set one nonempty NEWSLENS_RELEASE_ROOT or NEWSLENS_ARTIFACT_PATH." + ) + from newslens.operations.lifecycle import serving_path + + raw_artifact_path = str(serving_path(Path(release_root).expanduser())) + raw_database_url = os.getenv(REALTIME_DATABASE_URL_ENVIRONMENT_VARIABLE) raw_candidate_limit = os.getenv(REALTIME_CANDIDATE_LIMIT_ENVIRONMENT_VARIABLE) diff --git a/src/newslens/operations/__init__.py b/src/newslens/operations/__init__.py new file mode 100644 index 0000000..e91638a --- /dev/null +++ b/src/newslens/operations/__init__.py @@ -0,0 +1 @@ +"""Repeatable offline training and explicit release management.""" diff --git a/src/newslens/operations/lifecycle.py b/src/newslens/operations/lifecycle.py new file mode 100644 index 0000000..245cae0 --- /dev/null +++ b/src/newslens/operations/lifecycle.py @@ -0,0 +1,161 @@ +"""Content-addressed training runs and atomic local release pointers (POSIX).""" + +from __future__ import annotations + +import fcntl +import hashlib +import json +import math +import os +import tempfile +from contextlib import contextmanager +from pathlib import Path + +import pandas as pd + +from newslens.artifacts import export_fallback_artifact, load_artifact +from newslens.data import audit_dataset, load_behaviors, load_news +from newslens.evaluation.fallback import evaluate_fallback_baseline + + +def digest(path: Path) -> str: + with path.open("rb") as stream: + return hashlib.file_digest(stream, "sha256").hexdigest() + + +def atomic_json(path: Path, payload: dict) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + fd, temporary = tempfile.mkstemp(dir=path.parent, prefix=".write-") + try: + with os.fdopen(fd, "w") as stream: + json.dump(payload, stream, indent=2, allow_nan=False) + stream.flush() + os.fsync(stream.fileno()) + os.replace(temporary, path) + finally: + if os.path.exists(temporary): + os.unlink(temporary) + + +@contextmanager +def locked(root: Path): + root.mkdir(parents=True, exist_ok=True) + with (root / ".lifecycle.lock").open("a") as stream: + fcntl.flock(stream, fcntl.LOCK_EX) + yield + + +def train( + snapshot: Path, + root: Path, + *, + cutoff: str, + minimum_ndcg: float, + minimum_validation: int = 100, + max_features: int = 50000, +) -> dict: + """Evaluate the recipe chronologically, then refit on the available snapshot. + + The gate evaluates the training recipe, not independent quality of the refit. + This snapshot must exclude the official final holdout. Nothing is auto-promoted. + """ + if not math.isfinite(minimum_ndcg) or not 0 <= minimum_ndcg <= 1: + raise ValueError("minimum_ndcg must be finite and between zero and one") + if minimum_validation < 1: + raise ValueError("minimum_validation must be positive") + cutoff_time = pd.Timestamp(cutoff) + if pd.isna(cutoff_time): + raise ValueError("cutoff required") + sources = {name: digest(snapshot / name) for name in ("news.tsv", "behaviors.tsv")} + specification = { + "sources": sources, + "cutoff": cutoff_time.isoformat(), + "minimum_ndcg": minimum_ndcg, + "minimum_validation": minimum_validation, + "max_features": max_features, + "pipeline_version": 1, + } + run_id = hashlib.sha256(json.dumps(specification, sort_keys=True).encode()).hexdigest()[:24] + directory = root / "runs" / run_id + with locked(root): + receipt_path = directory / "receipt.json" + if receipt_path.exists(): + receipt = json.loads(receipt_path.read_text()) + if receipt["status"] == "candidate": + load_artifact(directory / "artifact") + if digest(directory / "artifact/manifest.json") != receipt["manifest_sha256"]: + raise ValueError("candidate manifest changed") + return receipt + news = load_news(snapshot / "news.tsv") + behaviors = load_behaviors(snapshot / "behaviors.tsv") + behaviors = behaviors.loc[behaviors.timestamp <= cutoff_time].copy() + if behaviors.empty: + raise ValueError("no behaviors at or before cutoff") + audit = audit_dataset(news, behaviors, "training-snapshot").to_dict() + if audit["referenced_news_missing_metadata"] or audit["missing_titles"]: + raise ValueError("snapshot failed catalog validation") + report = evaluate_fallback_baseline( + news, behaviors, max_features=max_features, bootstrap_samples=100 + ) + metrics = report.metrics.to_dict() + passed = ( + metrics["evaluated_impressions"] >= minimum_validation + and metrics["ndcg_at_k"] >= minimum_ndcg + and metrics["empty_ranking_impressions"] == 0 + ) + # Detect a concurrently replaced source before publishing any candidate. + if sources != {name: digest(snapshot / name) for name in sources}: + raise ValueError("snapshot changed during training") + receipt = { + "run_id": run_id, + "specification": specification, + "audit": audit, + "validation": metrics, + "status": "candidate" if passed else "rejected", + "evaluation_scope": "chronological recipe validation; refit is not final-test evidence", + } + directory.mkdir(parents=True, exist_ok=True) + if passed: + artifact = directory / "artifact" + # An interrupted export can leave a complete artifact before its receipt. + if artifact.exists(): + load_artifact(artifact) + else: + export_fallback_artifact(news, behaviors, artifact, max_features=max_features) + receipt["manifest_sha256"] = digest(artifact / "manifest.json") + atomic_json(receipt_path, receipt) + return receipt + + +def checked_candidate(root: Path, run_id: str) -> Path: + if len(run_id) != 24 or any(c not in "0123456789abcdef" for c in run_id): + raise ValueError("invalid run ID") + directory = root / "runs" / run_id + receipt = json.loads((directory / "receipt.json").read_text()) + if receipt.get("status") != "candidate" or receipt.get("run_id") != run_id: + raise ValueError("only a passing candidate can be released") + artifact = directory / "artifact" + if digest(artifact / "manifest.json") != receipt.get("manifest_sha256"): + raise ValueError("candidate manifest changed") + load_artifact(artifact) + return artifact.resolve() + + +def release(root: Path, run_id: str | None = None, *, rollback: bool = False) -> dict: + with locked(root): + pointer = root / "release.json" + state = json.loads(pointer.read_text()) if pointer.exists() else {} + selected = state.get("previous") if rollback else run_id + if not selected: + raise ValueError("no candidate/previous release selected") + checked_candidate(root, selected) + if state.get("current") == selected: + return state + result = {"current": selected, "previous": state.get("current")} + atomic_json(pointer, result) + return result + + +def serving_path(root: Path) -> Path: + state = json.loads((root / "release.json").read_text()) + return checked_candidate(root, state["current"]) diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py new file mode 100644 index 0000000..b8d50bf --- /dev/null +++ b/tests/test_lifecycle.py @@ -0,0 +1,94 @@ +import json + +import pytest +from test_artifact_export import make_behaviors, make_news + +from newslens.artifacts import export_fallback_artifact +from newslens.operations.lifecycle import atomic_json, digest, release, serving_path + + +def candidate(root, run_id): + path = root / "runs" / run_id + export_fallback_artifact(make_news(), make_behaviors(), path / "artifact") + atomic_json( + path / "receipt.json", + { + "run_id": run_id, + "status": "candidate", + "manifest_sha256": digest(path / "artifact/manifest.json"), + }, + ) + return path + + +def test_promote_rollback_and_corruption(tmp_path): + a, b = "a" * 24, "b" * 24 + candidate(tmp_path, a) + path = candidate(tmp_path, b) + release(tmp_path, a) + release(tmp_path, b) + assert serving_path(tmp_path) == (path / "artifact").resolve() + assert release(tmp_path, rollback=True)["current"] == a + (path / "artifact/model.joblib").write_bytes(b"corrupt") + with pytest.raises(RuntimeError): + release(tmp_path, b) + assert json.loads((tmp_path / "release.json").read_text())["current"] == a + + +def test_rejected_candidate_cannot_promote(tmp_path): + atomic_json(tmp_path / "runs" / ("a" * 24) / "receipt.json", {"status": "rejected"}) + with pytest.raises(ValueError, match="passing candidate"): + release(tmp_path, "a" * 24) + + +def test_training_retry_cutoff_and_api_release(tmp_path, monkeypatch): + from newslens.api.settings import ApiSettings, ApiSettingsError + from newslens.operations.lifecycle import train + + snapshot = tmp_path / "snapshot" + snapshot.mkdir() + (snapshot / "news.tsv").write_text( + "N1\tscience\tspace\tMars water\tPlanet exploration\turl\t[]\t[]\nN2\tsport\tball\tMars team\tChampionship wins\turl\t[]\t[]\n" + ) + (snapshot / "behaviors.tsv").write_text( + "".join(f"{i}\tU1\t01/{i:02d}/2020 12:00:00 AM\tN1\tN1-0 N2-1\n" for i in range(1, 11)) + ) + root = tmp_path / "registry" + args = {"cutoff": "2020-01-09", "minimum_ndcg": 0, "minimum_validation": 1} + result = train(snapshot, root, **args) + assert result["status"] == "candidate" + assert train(snapshot, root, **args) == result + release(root, result["run_id"]) + monkeypatch.delenv("NEWSLENS_ARTIFACT_PATH", raising=False) + monkeypatch.setenv("NEWSLENS_RELEASE_ROOT", str(root)) + assert ApiSettings.from_environment().artifact_path == serving_path(root) + monkeypatch.setenv("NEWSLENS_ARTIFACT_PATH", "other") + with pytest.raises(ApiSettingsError): + ApiSettings.from_environment() + with pytest.raises(ValueError, match="no behaviors"): + train(snapshot, root, **{**args, "cutoff": "1999-01-01"}) + + +def test_rehearsal_refuses_live_registry(tmp_path): + from scripts.rehearse_lifecycle import rehearse + + with pytest.raises(ValueError, match="isolated registry"): + rehearse(tmp_path, tmp_path, ["2020-01-01", "2020-01-02"], 0, 1) + + +def test_complete_release_rehearsal(tmp_path): + from scripts.rehearse_lifecycle import rehearse + + snapshot = tmp_path / "snapshot" + snapshot.mkdir() + (snapshot / "news.tsv").write_text( + "N1\tscience\tspace\tMars water\tPlanet\turl\t[]\t[]\n" + "N2\tscience\tspace\tMars team\tPlanet\turl\t[]\t[]\n" + ) + (snapshot / "behaviors.tsv").write_text( + "".join(f"{i}\tU1\t01/{i:02d}/2020 12:00:00 AM\tN1\tN1-0 N2-1\n" for i in range(1, 11)) + ) + result = rehearse(snapshot, tmp_path / "registry", ["2020-01-08", "2020-01-09"], 0, 1) + assert result["status"] == "completed" + assert [e["step"] for e in result["events"]][-1] == "rollback_and_resolve" + assert result["events"][-1]["run_id"] == result["events"][0]["run_id"]