Add verified model training and release lifecycle

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).

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.

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()

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()

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)
 

src/newslens/operations/__init__.py

@@ -0,0 +1 @@
+"""Repeatable offline training and explicit release management."""

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"])