From db0adad8d6af4a73a81ae93889570850b8e90cd0 Mon Sep 17 00:00:00 2001 From: appleweiping Date: Sun, 20 Sep 2026 02:53:44 -0500 Subject: [PATCH 1/2] Add bounded trainable keyphrase centroid expertise Signed-off-by: appleweiping --- .github/workflows/ci.yml | 17 + CHANGELOG.md | 9 + CITATION.cff | 2 +- README.md | 9 +- docs/trainable-centroid.md | 113 +++ examples/centroid/documents.json | 5 + examples/centroid/experts.json | 4 + examples/centroid/train.jsonl | 1 + examples/centroid/validation.jsonl | 1 + pyproject.toml | 2 +- src/peermatchlab/__init__.py | 24 +- src/peermatchlab/cli.py | 48 ++ src/peermatchlab/keyphrase_centroid.py | 952 +++++++++++++++++++++++++ tests/test_keyphrase_centroid.py | 456 ++++++++++++ uv.lock | 2 +- 15 files changed, 1640 insertions(+), 5 deletions(-) create mode 100644 docs/trainable-centroid.md create mode 100644 examples/centroid/documents.json create mode 100644 examples/centroid/experts.json create mode 100644 examples/centroid/train.jsonl create mode 100644 examples/centroid/validation.jsonl create mode 100644 src/peermatchlab/keyphrase_centroid.py create mode 100644 tests/test_keyphrase_centroid.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 737b0a3..29d60f7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -43,6 +43,22 @@ jobs: uv run --locked peermatch extract-keyphrases --documents examples/documents.json --experts examples/experts.json --directory keyphrases-smoke + - run: >- + uv run --locked peermatch extract-keyphrases + --documents examples/centroid/documents.json --experts examples/centroid/experts.json + --directory centroid-keyphrases-smoke + - run: >- + uv run --locked peermatch expertise-centroid + --keyphrases centroid-keyphrases-smoke + --documents examples/centroid/documents.json --experts examples/centroid/experts.json + --train examples/centroid/train.jsonl --validation examples/centroid/validation.jsonl + --dimensions 2 --epochs 3 --directory centroid-run-smoke + - run: >- + uv run --locked peermatch match-affinity + --documents centroid-run-smoke/documents.json --experts centroid-run-smoke/experts.json + --conflicts centroid-run-smoke/conflicts.json + --affinities centroid-run-smoke/affinities.csv + --config examples/expertise/match-config.json --output centroid-plan-smoke.json - run: >- uv run --locked peermatch expertise-embedding --snapshot examples/expertise/snapshot @@ -81,6 +97,7 @@ jobs: .ci-wheel-check/bin/python -m pip install --disable-pip-version-check --no-deps dist/*.whl .ci-wheel-check/bin/python -c "import peermatchlab; print(peermatchlab.__version__)" .ci-wheel-check/bin/python -c "from peermatchlab.keyphrases import KeyphraseConfig, rank_keyphrases; assert rank_keyphrases('alpha beta', config=KeyphraseConfig())" + .ci-wheel-check/bin/python -c "from peermatchlab import CentroidConfig, train_keyphrase_centroid; assert CentroidConfig and train_keyphrase_centroid" .ci-wheel-check/bin/python -c "from peermatchlab.what_if import WhatIfPlan, compare_what_if; assert WhatIfPlan and compare_what_if" .ci-wheel-check/bin/peermatch --help diff --git a/CHANGELOG.md b/CHANGELOG.md index 979f8cc..ef02093 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,15 @@ All notable changes are documented here. The project follows semantic versioning _No changes yet._ +## 0.16.0 - 2026-09-20 + +- Added an original, bounded local trainable keyphrase-centroid expertise + workflow: strict hashed TextRank-style source interchange, train/validation + triplets split by submission ID, deterministic pairwise logistic SGD, + first-best validation MAP checkpoint, replayable source-bound model, and + holdout-only affinities consumable by `match-affinity`. It uses synthetic + fixtures and does not claim OpenReview Expertise model/score parity. + ## 0.15.0 - 2026-09-20 - Added bounded, deterministic local capacity/conflict what-if comparisons with diff --git a/CITATION.cff b/CITATION.cff index 45c16bc..f4e962b 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -6,7 +6,7 @@ authors: - family-names: "Yan" given-names: "Weiping" alias: "appleweiping" -version: 0.15.0 +version: 0.16.0 date-released: 2026-09-20 license: MIT repository-code: "https://github.com/appleweiping/PeerMatchLab" diff --git a/README.md b/README.md index 30c074c..8a564fb 100644 --- a/README.md +++ b/README.md @@ -27,6 +27,9 @@ Most matching prototypes stop after computing pairwise similarity. Real allocati publication/profile evidence and additive term explanations. - Offline, pluggable SPECTER-family embedding interchange with hashed synthetic fixtures; no neural encoder or model weights are bundled. +- Bounded [trainable keyphrase-centroid expertise](docs/trainable-centroid.md) with + disjoint train/validation submission IDs, pairwise logistic updates, + first-best validation MAP checkpoint, and holdout-only match-ready affinities. - Content similarity, explicit topic overlap, bid preference, publication recency, and seniority components. - Exact hard-conflict and zero-capacity exclusion before optimization. - Integral min-cost-flow assignment that maximizes total score. @@ -371,7 +374,11 @@ the frozen comparator's cosine, global normalization, and max/average reviewer scoring boundary without claiming that the included synthetic vectors came from SPECTER. See the [embedding interchange contract](docs/embedding-expertise.md). An opt-in `--aggregation centroid` averages individually normalized local -publication vectors; it is not the frozen trainable keyphrase-centroid model. +publication vectors; it does not train keyphrase vectors. The separate +`peermatch expertise-centroid` command does train local keyphrase vectors from +caller-authored triplets and emits holdout affinities for `match-affinity`. +See the [bounded training protocol](docs/trainable-centroid.md); neither mode +claims numerical or dataset parity with the frozen OpenReview model. To evaluate those sparse affinities against a permitted local gold standard, use `peermatch evaluate-gold`. It supports canonical judged triples and the diff --git a/docs/trainable-centroid.md b/docs/trainable-centroid.md new file mode 100644 index 0000000..c3ec3eb --- /dev/null +++ b/docs/trainable-centroid.md @@ -0,0 +1,113 @@ +# Bounded trainable keyphrase-centroid expertise + +`peermatch expertise-centroid` is an original, dependency-free offline +training/inference workflow. It consumes a **previously generated** +[`extract-keyphrases` run](keyphrases.md), the exact document/expert source +bytes named in that run, and caller-authored train/validation triplet JSONL. +It does not download data, call a provider, run PyTorch or the frozen OpenReview +Expertise model, or reproduce its official gold-data scores. + +## Synthetic end-to-end run + +These files are fictional. From the repository root: + +```bash +peermatch extract-keyphrases \ + --documents examples/centroid/documents.json \ + --experts examples/centroid/experts.json \ + --directory centroid-keyphrases + +peermatch expertise-centroid \ + --keyphrases centroid-keyphrases \ + --documents examples/centroid/documents.json \ + --experts examples/centroid/experts.json \ + --train examples/centroid/train.jsonl \ + --validation examples/centroid/validation.jsonl \ + --dimensions 2 --epochs 3 --seed 17 \ + --directory centroid-run + +peermatch match-affinity \ + --documents centroid-run/documents.json \ + --experts centroid-run/experts.json \ + --conflicts centroid-run/conflicts.json \ + --affinities centroid-run/affinities.csv \ + --config examples/expertise/match-config.json \ + --output centroid-assignment.json +``` + +`--conflicts` can be supplied during centroid preparation. A positive +training/validation pair that is a declared hard conflict is rejected; the +holdout conflict subset is copied to the run for assignment. Inference emits +scores for all holdout document/reviewer pairs; `match-affinity` still applies +the hard constraints. Neither a high score nor a training label overrides a +declared conflict, capacity, seniority, or institution rule. + +Both output directories are create-only. Choose new names for reruns; neither +command overwrites existing files. The run contains a versioned `model.json`, +`affinities.csv`, normalized **holdout-only** `documents.json`, `experts.json`, +`conflicts.json`, and an aggregate hash/byte/count `manifest.json`. The +checkpoint stores a learned vocabulary and weights, so a real venue must +handle it as private derived data; the manifest does not include raw text or +labels. Preserve the original inputs to replay or audit the declared hashes. + +## Data separation and scoring + +Each JSONL row is exactly: + +```json +{"document_id":"d-train","positive_expert_id":"e-math","negative_expert_id":"e-graph"} +``` + +Train and validation must contain **disjoint submission IDs**. Repeated +triplets, contradictory labels for one submission/reviewer pair, missing +submission or reviewer keyphrases, and unknown IDs fail closed. Every source +byte stream is capped and SHA-256-bound to the keyphrase extraction manifest +or checkpoint. Vocabulary is formed only from training submissions and +reviewers occurring in training triplets. Reviewer evidence combines its +profile and publication keyphrases, deduplicating terms. Reviewer text may be +shared as *unlabeled features* across partitions; validation labels never +enter weight updates. A validation or holdout entity with no known term is +rejected rather than scored with untrained random coordinates. All +submission IDs absent from both label partitions are holdout scoring inputs. +The tool does **not** infer that this split is unbiased or temporally safe; +the caller must build partitions before examining holdout outcomes and avoid +source/author leakage across their own data. + +For a document `q` and reviewer `r`, take the arithmetic mean of their +learned term vectors and score their dot product. A training triplet has +margin `q·positive - q·negative` and loss `log(1 + exp(-margin))`. +Synchronous, deterministic SGD differentiates through each centroid and +applies optional L2 weight decay to touched term vectors. The seed controls +local uniform initialization; training order is sorted by IDs. At the end of +each epoch, the untouched validation labels evaluate per-submission average +precision over *judged* positive/negative reviewers, with reviewer-ID tie +breaking. The first epoch attaining the highest validation MAP is selected; +validation never changes weights. Raw holdout dot products pass through a +logistic transform and are rounded to six decimals to satisfy the existing +`[0,1]` affinity contract. These numbers are **not calibrated probabilities**. + +## Finite bounds and audit + +Input files: keyphrase/document/expert sources at most 16 MiB each; +train/validation/conflicts at most 2 MiB each; keyphrase manifest 64 KiB. +At most 10,000 keyphrase records, 2,000 train and 500 validation triplets, +4,096 learned terms, 256 terms per entity, 32 dimensions, 50 epochs, and +10,000 holdout/reviewer affinities are accepted. A configured coordinate-work +bound defaults to 20 million and covers training, validation, and inference; +generated model and run byte bounds are independently enforced. Empty/missing +evidence, nonfinite coordinates/loss, invalid dimensions/IDs, forged source +hashes, or a changed input path before publication are errors. + +`load_centroid_model` verifies the checkpoint's canonical payload digest and +shape; `verify_keyphrase_centroid` refits from the exact captured inputs and +compares checkpoint bytes. The run writer performs that replay and rechecks +source paths before publishing a staged directory with no replacement. An +independent hand calculation in the tests checks one pairwise BCE gradient +step; a separate two-submission ranking oracle checks MAP. Real-world use +still requires representative labels, licensing, bias/contamination review, +held-out evaluation, declared conflicts, and human oversight. + +The frozen OpenReview Expertise centroid uses a different PyTorch training, +optimizer, batcher, keyphrase and evaluation pipeline. This slice closes a +local trainable-keyphrase outcome, **not** full comparator parity, TPMS, +multifacet neural scoring, SPECTER inference, or official gold evaluation. diff --git a/examples/centroid/documents.json b/examples/centroid/documents.json new file mode 100644 index 0000000..1bd0c2e --- /dev/null +++ b/examples/centroid/documents.json @@ -0,0 +1,5 @@ +[ + {"id":"d-train","title":"Algebra matrix proof","abstract":"Linear algebra matrix theorem","required_experts":1}, + {"id":"d-val","title":"Matrix algebra methods","abstract":"Linear matrix algebra","required_experts":1}, + {"id":"d-hold","title":"Algebra matrix applications","abstract":"Matrix proof algebra","required_experts":1} +] diff --git a/examples/centroid/experts.json b/examples/centroid/experts.json new file mode 100644 index 0000000..d5b90e1 --- /dev/null +++ b/examples/centroid/experts.json @@ -0,0 +1,4 @@ +[ + {"id":"e-math","name":"Math reviewer","summary":"Algebra matrix theorem","capacity":1}, + {"id":"e-graph","name":"Graph reviewer","summary":"Graph networks circuits","capacity":1} +] diff --git a/examples/centroid/train.jsonl b/examples/centroid/train.jsonl new file mode 100644 index 0000000..e04fab8 --- /dev/null +++ b/examples/centroid/train.jsonl @@ -0,0 +1 @@ +{"document_id":"d-train","positive_expert_id":"e-math","negative_expert_id":"e-graph"} diff --git a/examples/centroid/validation.jsonl b/examples/centroid/validation.jsonl new file mode 100644 index 0000000..b46da88 --- /dev/null +++ b/examples/centroid/validation.jsonl @@ -0,0 +1 @@ +{"document_id":"d-val","positive_expert_id":"e-math","negative_expert_id":"e-graph"} diff --git a/pyproject.toml b/pyproject.toml index 7145c20..199142f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "peermatchlab" -version = "0.15.0" +version = "0.16.0" description = "Transparent, constraint-aware expert matching for offline workflows" readme = "README.md" requires-python = ">=3.11" diff --git a/src/peermatchlab/__init__.py b/src/peermatchlab/__init__.py index abfdfae..aab1f9a 100644 --- a/src/peermatchlab/__init__.py +++ b/src/peermatchlab/__init__.py @@ -34,6 +34,18 @@ local_domain_inputs, write_expertise_run, ) +from peermatchlab.keyphrase_centroid import ( + CentroidConfig, + CentroidInputs, + CentroidModel, + CentroidTriplet, + load_centroid_inputs, + load_centroid_model, + score_keyphrase_centroid, + train_keyphrase_centroid, + verify_keyphrase_centroid, + write_keyphrase_centroid_run, +) from peermatchlab.keyphrases import ( Keyphrase, KeyphraseConfig, @@ -98,6 +110,10 @@ "AssignmentEngine", "AssignmentStrategy", "AssignmentWriteTransport", + "CentroidConfig", + "CentroidInputs", + "CentroidModel", + "CentroidTriplet", "DemandDiagnostic", "Document", "EmbeddingProvider", @@ -149,6 +165,8 @@ "fetch_reviewer_expertise", "generate_expertise", "load_affinities_csv", + "load_centroid_inputs", + "load_centroid_model", "load_expertise_config_source", "load_local_domain_expertise_inputs", "load_openreview_expertise_snapshot", @@ -163,11 +181,15 @@ "request_sha256", "reviewer_ids_to_experts", "score_embedding_expertise", + "score_keyphrase_centroid", + "train_keyphrase_centroid", + "verify_keyphrase_centroid", "write_embedding_run", "write_expertise_run", + "write_keyphrase_centroid_run", "write_keyphrase_run", "write_openreview_snapshot", "write_publish_artifact", ] -__version__ = "0.15.0" +__version__ = "0.16.0" diff --git a/src/peermatchlab/cli.py b/src/peermatchlab/cli.py index 2467515..d0f230c 100644 --- a/src/peermatchlab/cli.py +++ b/src/peermatchlab/cli.py @@ -38,6 +38,12 @@ plan_to_dict, write_json, ) +from peermatchlab.keyphrase_centroid import ( + CentroidConfig, + load_centroid_inputs, + train_keyphrase_centroid, + write_keyphrase_centroid_run, +) from peermatchlab.keyphrases import KeyphraseConfig, read_keyphrase_source, write_keyphrase_run from peermatchlab.models import Conflict, DataValidationError, Document, Expert from peermatchlab.openreview import load_openreview_submissions, load_reviewer_ids @@ -228,6 +234,22 @@ def build_parser() -> argparse.ArgumentParser: keyphrases.add_argument("--window-size", type=int, default=2) keyphrases.add_argument("--iterations", type=int, default=30) keyphrases.add_argument("--damping", type=float, default=0.85) + centroid = commands.add_parser( + "expertise-centroid", help="train and score a bounded local keyphrase centroid" + ) + centroid.add_argument("--keyphrases", required=True, help="extract-keyphrases run directory") + centroid.add_argument("--documents", required=True, help="same exact document source bytes") + centroid.add_argument("--experts", required=True, help="same exact expert source bytes") + centroid.add_argument("--train", required=True, help="train triplets JSONL") + centroid.add_argument("--validation", required=True, help="disjoint validation triplets JSONL") + centroid.add_argument("--conflicts", help="optional hard-conflict JSON/JSONL") + centroid.add_argument("--dimensions", type=int, default=8) + centroid.add_argument("--epochs", type=int, default=8) + centroid.add_argument("--learning-rate", type=float, default=0.1) + centroid.add_argument("--l2", type=float, default=0.0) + centroid.add_argument("--seed", type=int, default=17) + centroid.add_argument("--max-work", type=int, default=20_000_000) + centroid.add_argument("--directory", required=True, help="new match-ready artifact directory") return parser @@ -609,6 +631,32 @@ def main(argv: Sequence[str] | None = None) -> int: ) print(f"extracted {manifest['records']} evidence records to {args.directory}") return 0 + if args.command == "expertise-centroid": + inputs = load_centroid_inputs( + args.keyphrases, + args.documents, + args.experts, + args.train, + args.validation, + conflicts_path=args.conflicts, + ) + model = train_keyphrase_centroid( + inputs, + config=CentroidConfig( + dimensions=args.dimensions, + epochs=args.epochs, + learning_rate=args.learning_rate, + l2=args.l2, + seed=args.seed, + max_work=args.max_work, + ), + ) + manifest = write_keyphrase_centroid_run(inputs, model, args.directory) + counts = manifest["counts"] + if not isinstance(counts, Mapping): + raise DataValidationError("centroid manifest counts are invalid") + print(f"generated {counts['affinities']} holdout affinities to {args.directory}") + return 0 if args.command == "evaluate-gold": eval_report = evaluate_gold_files( args.gold, diff --git a/src/peermatchlab/keyphrase_centroid.py b/src/peermatchlab/keyphrase_centroid.py new file mode 100644 index 0000000..46be3d5 --- /dev/null +++ b/src/peermatchlab/keyphrase_centroid.py @@ -0,0 +1,952 @@ +"""Offline, bounded pairwise training of mean-keyphrase reviewer centroids. + +This is an original, dependency-free local protocol. It does not run the frozen +OpenReview Expertise PyTorch model or claim its checkpoint/score equivalence. +""" + +from __future__ import annotations + +import csv +import hashlib +import io +import json +import math +import random +import tempfile +from collections import defaultdict +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType +from typing import Any + +from peermatchlab.affinity import Affinity +from peermatchlab.expertise_io import _install_directory_no_replace +from peermatchlab.io import ( + load_conflicts_text, + load_documents_text, + load_experts_text, + load_json_text, +) +from peermatchlab.models import ( + Conflict, + DataValidationError, + Document, + Expert, + _identifier_is_valid, +) + +PROTOCOL = "peermatch-keyphrase-centroid-v1" +_MAX_SOURCE = 16 * 1024 * 1024 +_MAX_LABEL_SOURCE = 2 * 1024 * 1024 +_MAX_MODEL = 8 * 1024 * 1024 +_MAX_TERMS = 4096 +_MAX_ENTITY_TERMS = 256 +_MAX_PAIRS = 10_000 +_MAX_WORK = 20_000_000 + + +def _json(value: object) -> bytes: + try: + return ( + json.dumps( + value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False + ) + + "\n" + ).encode("utf-8", "strict") + except (TypeError, ValueError, UnicodeError, RecursionError) as error: + raise DataValidationError("centroid artifact cannot be encoded as strict JSON") from error + + +def _sha(raw: bytes) -> str: + return hashlib.sha256(raw).hexdigest() + + +def _read(path: str | Path, maximum: int, label: str) -> bytes: + try: + with Path(path).open("rb") as stream: + raw = stream.read(maximum + 1) + except OSError as error: + raise DataValidationError(f"cannot read {label}") from error + if len(raw) > maximum: + raise DataValidationError(f"{label} exceeds {maximum} bytes") + return raw + + +def _decode(raw: bytes, label: str) -> str: + try: + return raw.decode("utf-8", "strict") + except UnicodeError as error: + raise DataValidationError(f"{label} must be strict UTF-8") from error + + +def _json_lines(raw: bytes, label: str, maximum: int) -> tuple[dict[str, Any], ...]: + rows: list[dict[str, Any]] = [] + for number, line in enumerate(_decode(raw, label).splitlines(), 1): + if not line or len(line.encode("utf-8")) > 64 * 1024: + raise DataValidationError(f"{label} has blank or oversized line {number}") + if len(rows) >= maximum: + raise DataValidationError(f"{label} exceeds {maximum} rows") + try: + value = load_json_text(line) + except (ValueError, RecursionError) as error: + raise DataValidationError(f"invalid {label} line {number}") from error + if not isinstance(value, dict): + raise DataValidationError(f"{label} line {number} must be an object") + rows.append(value) + if not rows: + raise DataValidationError(f"{label} is empty") + return tuple(rows) + + +def _finite_real(value: object) -> bool: + if not isinstance(value, (int, float)) or type(value) not in (int, float): + return False + try: + return math.isfinite(float(value)) + except (OverflowError, ValueError): + return False + + +@dataclass(frozen=True, slots=True) +class CentroidConfig: + """Hard ceilings for one deterministic pairwise-SGD fit.""" + + dimensions: int = 8 + epochs: int = 8 + learning_rate: float = 0.1 + l2: float = 0.0 + seed: int = 17 + max_work: int = _MAX_WORK + + def __post_init__(self) -> None: + for name, low, high in ( + ("dimensions", 1, 32), + ("epochs", 1, 50), + ("seed", 0, 2**32 - 1), + ("max_work", 1, _MAX_WORK), + ): + value = getattr(self, name) + if type(value) is not int or not low <= value <= high: + raise DataValidationError(f"{name} must be an integer in [{low}, {high}]") + for name, lower_float, upper_float in (("learning_rate", 0.0, 1.0), ("l2", 0.0, 1.0)): + float_value = getattr(self, name) + if not _finite_real(float_value) or not lower_float <= float_value <= upper_float: + raise DataValidationError( + f"{name} must be finite in [{lower_float}, {upper_float}]" + ) + if self.learning_rate == 0: + raise DataValidationError("learning_rate must be positive") + + def as_dict(self) -> dict[str, int | float]: + return {name: getattr(self, name) for name in self.__dataclass_fields__} + + +@dataclass(frozen=True, slots=True) +class CentroidTriplet: + document_id: str + positive_expert_id: str + negative_expert_id: str + + +def _triplets(raw: bytes, label: str, maximum: int) -> tuple[CentroidTriplet, ...]: + rows = _json_lines(raw, label, maximum) + result: list[CentroidTriplet] = [] + seen: set[CentroidTriplet] = set() + labels: dict[tuple[str, str], bool] = {} + for row in rows: + if set(row) != {"document_id", "positive_expert_id", "negative_expert_id"}: + raise DataValidationError(f"{label} triplet schema is invalid") + item = CentroidTriplet( + row["document_id"], row["positive_expert_id"], row["negative_expert_id"] + ) + if ( + any( + not _identifier_is_valid(value) + for value in (item.document_id, item.positive_expert_id, item.negative_expert_id) + ) + or item.positive_expert_id == item.negative_expert_id + ): + raise DataValidationError(f"{label} triplet IDs must be valid and distinct") + if item in seen: + raise DataValidationError(f"{label} repeats a triplet") + seen.add(item) + for reviewer, positive in ( + (item.positive_expert_id, True), + (item.negative_expert_id, False), + ): + pair = (item.document_id, reviewer) + if pair in labels and labels[pair] is not positive: + raise DataValidationError(f"{label} has contradictory pair labels") + labels[pair] = positive + result.append(item) + return tuple( + sorted( + result, + key=lambda row: (row.document_id, row.positive_expert_id, row.negative_expert_id), + ) + ) + + +def _safe_term(value: object) -> bool: + if ( + type(value) is not str + or len(value) <= 1 + or not value.isalnum() + or value != value.casefold() + ): + return False + try: + return len(value.encode("utf-8", "strict")) <= 128 + except UnicodeError: + return False + + +def _safe_score(value: object) -> bool: + return _finite_real(value) and isinstance(value, (int, float)) and value >= 0 + + +def _keyphrase_terms( + raw: bytes, expected_count: int +) -> tuple[dict[str, tuple[str, ...]], dict[str, tuple[str, ...]]]: + records = _json_lines(raw, "keyphrase records", 10_000) + if len(records) != expected_count: + raise DataValidationError("keyphrase manifest record count mismatch") + seen: set[tuple[str, str, str]] = set() + submissions: dict[str, tuple[str, ...]] = {} + reviewers: dict[str, list[tuple[str, ...]]] = defaultdict(list) + profiles: set[str] = set() + for row in records: + if set(row) != {"kind", "owner_id", "evidence_id", "token_count", "keyphrases"}: + raise DataValidationError("keyphrase record schema is invalid") + kind, owner, evidence = row["kind"], row["owner_id"], row["evidence_id"] + if kind not in {"submission", "profile", "publication"} or not all( + _identifier_is_valid(value) for value in (owner, evidence) + ): + raise DataValidationError("keyphrase record identity is invalid") + identity = (kind, owner, evidence) + if identity in seen or (kind in {"submission", "profile"} and evidence != owner): + raise DataValidationError("keyphrase record identity is duplicate or inconsistent") + seen.add(identity) + if type(row["token_count"]) is not int or not 0 <= row["token_count"] <= 100_000: + raise DataValidationError("keyphrase token_count is invalid") + phrases = row["keyphrases"] + if not isinstance(phrases, list) or len(phrases) > 1000: + raise DataValidationError("keyphrase list exceeds limit") + terms: list[str] = [] + for phrase in phrases: + if not isinstance(phrase, dict) or set(phrase) != {"term", "score"}: + raise DataValidationError("keyphrase term schema is invalid") + term, score = phrase["term"], phrase["score"] + if not _safe_term(term) or not _safe_score(score) or term in terms: + raise DataValidationError("keyphrase term or score is invalid") + terms.append(term) + if kind == "submission": + submissions[owner] = tuple(terms) + else: + if kind == "profile": + profiles.add(owner) + reviewers[owner].append(tuple(terms)) + combined: dict[str, tuple[str, ...]] = {} + for owner, groups in reviewers.items(): + combined_terms = tuple(dict.fromkeys(term for group in groups for term in group)) + if len(combined_terms) > _MAX_ENTITY_TERMS: + raise DataValidationError("reviewer keyphrases exceed per-entity limit") + combined[owner] = combined_terms + if set(reviewers) != profiles: + raise DataValidationError("reviewer keyphrase profile records are missing") + return submissions, combined + + +@dataclass(frozen=True, slots=True) +class CentroidInputs: + documents: tuple[Document, ...] + experts: tuple[Expert, ...] + conflicts: tuple[Conflict, ...] + submissions: Mapping[str, tuple[str, ...]] + reviewers: Mapping[str, tuple[str, ...]] + train: tuple[CentroidTriplet, ...] + validation: tuple[CentroidTriplet, ...] + raw: Mapping[str, bytes] + paths: Mapping[str, Path] + + @property + def hashes(self) -> dict[str, str]: + return {name: _sha(data) for name, data in sorted(self.raw.items())} + + @property + def holdout_ids(self) -> tuple[str, ...]: + labelled = {item.document_id for item in (*self.train, *self.validation)} + return tuple(sorted(set(self.submissions) - labelled)) + + +def load_centroid_inputs( + keyphrases_directory: str | Path, + documents_path: str | Path, + experts_path: str | Path, + train_path: str | Path, + validation_path: str | Path, + *, + conflicts_path: str | Path | None = None, +) -> CentroidInputs: + """Load exact byte snapshots and disjoint, fully referenced label partitions.""" + paths = { + "keyphrases.jsonl": Path(keyphrases_directory) / "keyphrases.jsonl", + "keyphrase-manifest.json": Path(keyphrases_directory) / "manifest.json", + "documents": Path(documents_path), + "experts": Path(experts_path), + "train": Path(train_path), + "validation": Path(validation_path), + } + if conflicts_path is not None: + paths["conflicts"] = Path(conflicts_path) + if len({path.resolve(strict=False) for path in paths.values()}) != len(paths): + raise DataValidationError("centroid input paths must be distinct") + raw = { + name: _read( + path, + 64 * 1024 + if name == "keyphrase-manifest.json" + else _MAX_LABEL_SOURCE + if name in {"train", "validation", "conflicts"} + else _MAX_SOURCE, + name, + ) + for name, path in paths.items() + } + try: + manifest = load_json_text(_decode(raw["keyphrase-manifest.json"], "keyphrase manifest")) + except (ValueError, RecursionError) as error: + raise DataValidationError("keyphrase manifest is invalid") from error + if ( + not isinstance(manifest, dict) + or set(manifest) + != { + "schema_version", + "algorithm", + "config", + "records", + "documents_sha256", + "experts_sha256", + "keyphrases_sha256", + "keyphrases_bytes", + } + or type(manifest["schema_version"]) is not int + or manifest["schema_version"] != 1 + or manifest["algorithm"] != "lexical-undirected-pagerank-v1" + or not isinstance(manifest["config"], dict) + ): + raise DataValidationError("unsupported keyphrase manifest") + if ( + type(manifest["records"]) is not int + or not 1 <= manifest["records"] <= 10_000 + or type(manifest["keyphrases_bytes"]) is not int + or manifest["keyphrases_bytes"] != len(raw["keyphrases.jsonl"]) + or manifest["keyphrases_sha256"] != _sha(raw["keyphrases.jsonl"]) + or manifest["documents_sha256"] != _sha(raw["documents"]) + or manifest["experts_sha256"] != _sha(raw["experts"]) + ): + raise DataValidationError("keyphrase manifest source hashes/counts disagree") + documents = load_documents_text(_decode(raw["documents"], "documents"), max_records=10_000) + experts = load_experts_text(_decode(raw["experts"], "experts"), max_records=10_000) + conflicts = ( + load_conflicts_text(_decode(raw["conflicts"], "conflicts"), max_records=10_000) + if "conflicts" in raw + else () + ) + if len({(item.document_id, item.expert_id) for item in conflicts}) != len(conflicts): + raise DataValidationError("duplicate conflict pairs") + submissions, reviewers = _keyphrase_terms(raw["keyphrases.jsonl"], manifest["records"]) + if set(submissions) != {item.id for item in documents} or set(reviewers) != { + item.id for item in experts + }: + raise DataValidationError("keyphrase submission/reviewer IDs do not match source inputs") + if any(not terms or len(terms) > _MAX_ENTITY_TERMS for terms in submissions.values()) or any( + not terms for terms in reviewers.values() + ): + raise DataValidationError("submission/reviewer keyphrases are missing or exceed limit") + train = _triplets(raw["train"], "train", 2000) + validation = _triplets(raw["validation"], "validation", 500) + if {item.document_id for item in train} & {item.document_id for item in validation}: + raise DataValidationError("train and validation submission IDs overlap") + conflict_pairs = {(item.document_id, item.expert_id) for item in conflicts} + for item in (*train, *validation): + if ( + item.document_id not in submissions + or item.positive_expert_id not in reviewers + or item.negative_expert_id not in reviewers + ): + raise DataValidationError("triplet references missing keyphrase evidence") + if (item.document_id, item.positive_expert_id) in conflict_pairs: + raise DataValidationError("positive triplet pair is a declared hard conflict") + if any( + item.document_id not in submissions or item.expert_id not in reviewers for item in conflicts + ): + raise DataValidationError("conflict references unknown submission/reviewer") + result = CentroidInputs( + documents, + experts, + conflicts, + MappingProxyType(submissions), + MappingProxyType(reviewers), + train, + validation, + MappingProxyType(raw), + MappingProxyType(paths), + ) + if not result.holdout_ids: + raise DataValidationError( + "centroid scoring requires at least one unlabeled holdout submission" + ) + return result + + +def _centroid( + indices: tuple[int, ...], + weights: list[list[float]] | tuple[tuple[float, ...], ...], + dimensions: int, +) -> tuple[float, ...]: + if not indices: + raise DataValidationError("entity has no trained keyphrase coordinates") + return tuple( + math.fsum(weights[index][axis] for index in indices) / len(indices) + for axis in range(dimensions) + ) + + +def _sigmoid(value: float) -> float: + if value >= 0.0: + factor = math.exp(-value) + return 1.0 / (1.0 + factor) + factor = math.exp(value) + return factor / (1.0 + factor) + + +def _softplus_negative(margin: float) -> float: + return ( + math.log1p(math.exp(-margin)) if margin >= 0.0 else -margin + math.log1p(math.exp(margin)) + ) + + +def _dot(left: tuple[float, ...], right: tuple[float, ...]) -> float: + return math.fsum(a * b for a, b in zip(left, right, strict=True)) + + +def _update( + weights: list[list[float]], + query: tuple[int, ...], + positive: tuple[int, ...], + negative: tuple[int, ...], + *, + dimensions: int, + learning_rate: float, + l2: float, +) -> float: + """One exact synchronous gradient step for softplus(-q dot (p - n)).""" + q = _centroid(query, weights, dimensions) + p = _centroid(positive, weights, dimensions) + n = _centroid(negative, weights, dimensions) + margin = _dot(q, tuple(a - b for a, b in zip(p, n, strict=True))) + factor = _sigmoid(-margin) + gradients: dict[int, list[float]] = {} + for indices, vector, sign in ( + (query, tuple(a - b for a, b in zip(p, n, strict=True)), -1.0), + (positive, q, -1.0), + (negative, q, 1.0), + ): + for index in indices: + grad = gradients.setdefault(index, [0.0] * dimensions) + for axis in range(dimensions): + grad[axis] += sign * factor * vector[axis] / len(indices) + for index, grad in gradients.items(): + for axis in range(dimensions): + value = weights[index][axis] - learning_rate * (grad[axis] + l2 * weights[index][axis]) + if not math.isfinite(value) or abs(value) > 1e6: + raise DataValidationError("centroid training coordinate diverged") + weights[index][axis] = value + return _softplus_negative(margin) + + +def _indices(terms: tuple[str, ...], lookup: dict[str, int]) -> tuple[int, ...]: + result = tuple(lookup[term] for term in terms if term in lookup) + if not result: + raise DataValidationError("validation/holdout entity has no train-vocabulary keyphrases") + return result + + +def _validation_map( + rows: tuple[CentroidTriplet, ...], + documents: dict[str, tuple[int, ...]], + reviewers: dict[str, tuple[int, ...]], + weights: list[list[float]], + dimensions: int, +) -> float: + by_document: dict[str, dict[str, bool]] = defaultdict(dict) + for row in rows: + by_document[row.document_id][row.positive_expert_id] = True + by_document[row.document_id][row.negative_expert_id] = False + ap_values: list[float] = [] + for document_id in sorted(by_document): + q = _centroid(documents[document_id], weights, dimensions) + labels = by_document[document_id] + ranked = sorted( + labels, + key=lambda expert_id: ( + -_dot(q, _centroid(reviewers[expert_id], weights, dimensions)), + expert_id, + ), + ) + hits = 0 + precision_sum = 0.0 + for rank, expert_id in enumerate(ranked, 1): + if labels[expert_id]: + hits += 1 + precision_sum += hits / rank + ap_values.append(precision_sum / sum(labels.values())) + return math.fsum(ap_values) / len(ap_values) + + +@dataclass(frozen=True, slots=True) +class CentroidModel: + config: CentroidConfig + vocabulary: tuple[str, ...] + vectors: tuple[tuple[float, ...], ...] + best_epoch: int + validation_map: float + history: tuple[tuple[int, float, float], ...] + source_sha256: Mapping[str, str] + + def __post_init__(self) -> None: + if ( + not isinstance(self.config, CentroidConfig) + or not self.vocabulary + or len(self.vocabulary) > _MAX_TERMS + or any(not _safe_term(term) for term in self.vocabulary) + or tuple(sorted(set(self.vocabulary))) != self.vocabulary + or len(self.vectors) != len(self.vocabulary) + or any( + len(row) != self.config.dimensions + or any(not _finite_real(value) or abs(value) > 1e6 for value in row) + for row in self.vectors + ) + or type(self.best_epoch) is not int + or not 1 <= self.best_epoch <= self.config.epochs + or not _finite_real(self.validation_map) + or not 0 <= self.validation_map <= 1 + or len(self.history) != self.config.epochs + or any( + type(epoch) is not int + or epoch != index + or not _finite_real(loss) + or loss < 0 + or not _finite_real(metric) + or not 0 <= metric <= 1 + for index, (epoch, loss, metric) in enumerate(self.history, 1) + ) + or not self.source_sha256 + or set(self.source_sha256) + not in ( + { + "documents", + "experts", + "keyphrase-manifest.json", + "keyphrases.jsonl", + "train", + "validation", + }, + { + "documents", + "experts", + "keyphrase-manifest.json", + "keyphrases.jsonl", + "train", + "validation", + "conflicts", + }, + ) + or any( + type(value) is not str + or len(value) != 64 + or any(char not in "0123456789abcdef" for char in value) + for value in self.source_sha256.values() + ) + ): + raise DataValidationError( + "centroid model has invalid dimensions, coordinates or provenance" + ) + object.__setattr__(self, "source_sha256", MappingProxyType(dict(self.source_sha256))) + + def payload(self) -> dict[str, object]: + return { + "schema_version": 1, + "protocol": PROTOCOL, + "config": self.config.as_dict(), + "vocabulary": list(self.vocabulary), + "vectors": [list(row) for row in self.vectors], + "best_epoch": self.best_epoch, + "validation_map": self.validation_map, + "history": [ + {"epoch": epoch, "train_loss": loss, "validation_map": metric} + for epoch, loss, metric in self.history + ], + "source_sha256": dict(self.source_sha256), + } + + def to_bytes(self) -> bytes: + payload = self.payload() + raw = _json(payload) + if len(raw) > _MAX_MODEL: + raise DataValidationError("centroid checkpoint exceeds byte limit") + envelope = _json({"payload": payload, "sha256": _sha(raw)}) + if len(envelope) > _MAX_MODEL: + raise DataValidationError("centroid checkpoint exceeds byte limit") + return envelope + + +def load_centroid_model(raw: bytes) -> CentroidModel: + """Validate a versioned checkpoint and its canonical payload digest.""" + if type(raw) is not bytes or len(raw) > _MAX_MODEL: + raise DataValidationError("centroid checkpoint exceeds byte limit") + try: + envelope = load_json_text(_decode(raw, "centroid checkpoint")) + except (ValueError, RecursionError) as error: + raise DataValidationError("invalid centroid checkpoint JSON") from error + if ( + not isinstance(envelope, dict) + or set(envelope) != {"payload", "sha256"} + or not isinstance(envelope["payload"], dict) + ): + raise DataValidationError("centroid checkpoint envelope is invalid") + payload = envelope["payload"] + if ( + envelope["sha256"] != _sha(_json(payload)) + or set(payload) + != { + "schema_version", + "protocol", + "config", + "vocabulary", + "vectors", + "best_epoch", + "validation_map", + "history", + "source_sha256", + } + or payload["schema_version"] != 1 + or payload["protocol"] != PROTOCOL + ): + raise DataValidationError("centroid checkpoint digest or protocol is invalid") + try: + config = CentroidConfig(**payload["config"]) + history = tuple( + (row["epoch"], row["train_loss"], row["validation_map"]) for row in payload["history"] + ) + model = CentroidModel( + config, + tuple(payload["vocabulary"]), + tuple(tuple(row) for row in payload["vectors"]), + payload["best_epoch"], + payload["validation_map"], + history, + dict(payload["source_sha256"]), + ) + except (TypeError, KeyError, ValueError, AttributeError) as error: + raise DataValidationError("centroid checkpoint fields are invalid") from error + if model.to_bytes() != raw: + raise DataValidationError("centroid checkpoint is not canonical") + return model + + +def train_keyphrase_centroid( + inputs: CentroidInputs, *, config: CentroidConfig | None = None +) -> CentroidModel: + """Train on train triplets only; choose first highest-MAP validation epoch.""" + if not isinstance(inputs, CentroidInputs): + raise DataValidationError("centroid training requires validated inputs") + chosen = config or CentroidConfig() + train_docs = {row.document_id for row in inputs.train} + train_reviewers = { + reviewer + for row in inputs.train + for reviewer in (row.positive_expert_id, row.negative_expert_id) + } + vocabulary = tuple( + sorted( + {term for document_id in train_docs for term in inputs.submissions[document_id]} + | {term for reviewer_id in train_reviewers for term in inputs.reviewers[reviewer_id]} + ) + ) + if len(vocabulary) > _MAX_TERMS or len(vocabulary) * chosen.dimensions > 100_000: + raise DataValidationError("centroid vocabulary/dimension resource bound exceeded") + lookup = {term: index for index, term in enumerate(vocabulary)} + used_docs = ( + train_docs | {row.document_id for row in inputs.validation} | set(inputs.holdout_ids) + ) + used_reviewers = ( + train_reviewers + | { + reviewer + for row in inputs.validation + for reviewer in (row.positive_expert_id, row.negative_expert_id) + } + | set(inputs.reviewers) + ) + document_indices = {key: _indices(inputs.submissions[key], lookup) for key in used_docs} + reviewer_indices = {key: _indices(inputs.reviewers[key], lookup) for key in used_reviewers} + train_work = ( + sum( + len(document_indices[row.document_id]) + + len(reviewer_indices[row.positive_expert_id]) + + len(reviewer_indices[row.negative_expert_id]) + + 5 + for row in inputs.train + ) + * chosen.dimensions + * chosen.epochs + ) + validation_work = ( + sum( + len(document_indices[row.document_id]) + + len(reviewer_indices[row.positive_expert_id]) + + len(reviewer_indices[row.negative_expert_id]) + + 5 + for row in inputs.validation + ) + * chosen.dimensions + * chosen.epochs + ) + score_work = sum( + ( + len(document_indices[document_id]) + + len(reviewer_indices[reviewer_id]) + + chosen.dimensions + ) + * chosen.dimensions + for document_id in inputs.holdout_ids + for reviewer_id in inputs.reviewers + ) + if train_work + validation_work + score_work > chosen.max_work: + raise DataValidationError("centroid coordinate work exceeds max_work") + # Reproducible model initialization does not generate credentials or secrets. + rng = random.Random(chosen.seed) # nosec B311 + weights = [[rng.uniform(-0.1, 0.1) for _ in range(chosen.dimensions)] for _ in vocabulary] + best_map = -1.0 + best_epoch = 0 + best_vectors: tuple[tuple[float, ...], ...] = () + history: list[tuple[int, float, float]] = [] + for epoch in range(1, chosen.epochs + 1): + losses = [ + _update( + weights, + document_indices[row.document_id], + reviewer_indices[row.positive_expert_id], + reviewer_indices[row.negative_expert_id], + dimensions=chosen.dimensions, + learning_rate=chosen.learning_rate, + l2=chosen.l2, + ) + for row in inputs.train + ] + metric = _validation_map( + inputs.validation, document_indices, reviewer_indices, weights, chosen.dimensions + ) + loss = math.fsum(losses) / len(losses) + if not math.isfinite(loss): + raise DataValidationError("centroid training loss is non-finite") + history.append((epoch, loss, metric)) + if metric > best_map + 1e-12: + best_map, best_epoch = metric, epoch + best_vectors = tuple(tuple(row) for row in weights) + return CentroidModel( + chosen, vocabulary, best_vectors, best_epoch, best_map, tuple(history), inputs.hashes + ) + + +def score_keyphrase_centroid(inputs: CentroidInputs, model: CentroidModel) -> tuple[Affinity, ...]: + """Score only unseen holdout submissions for the existing assignment API.""" + if not isinstance(inputs, CentroidInputs) or not isinstance(model, CentroidModel): + raise DataValidationError("centroid scoring requires validated inputs and model") + if model.source_sha256 != inputs.hashes: + raise DataValidationError("centroid checkpoint source hashes do not match current inputs") + if len(inputs.holdout_ids) * len(inputs.experts) > _MAX_PAIRS: + raise DataValidationError("centroid affinity matrix exceeds pair limit") + lookup = {term: index for index, term in enumerate(model.vocabulary)} + document_indices = { + document_id: _indices(inputs.submissions[document_id], lookup) + for document_id in inputs.holdout_ids + } + reviewer_indices = { + expert.id: _indices(inputs.reviewers[expert.id], lookup) for expert in inputs.experts + } + work = sum( + ( + len(document_indices[document_id]) + + len(reviewer_indices[expert.id]) + + model.config.dimensions + ) + * model.config.dimensions + for document_id in inputs.holdout_ids + for expert in inputs.experts + ) + if work > model.config.max_work: + raise DataValidationError("centroid inference coordinate work exceeds max_work") + reviewer_centroids = { + expert.id: _centroid(reviewer_indices[expert.id], model.vectors, model.config.dimensions) + for expert in sorted(inputs.experts, key=lambda value: value.id) + } + result = [] + for document_id in inputs.holdout_ids: + query = _centroid( + document_indices[document_id], + model.vectors, + model.config.dimensions, + ) + for expert_id, centroid in reviewer_centroids.items(): + result.append( + Affinity(document_id, expert_id, round(_sigmoid(_dot(query, centroid)), 6)) + ) + return tuple(result) + + +def verify_keyphrase_centroid(inputs: CentroidInputs, checkpoint: bytes) -> bool: + """Replay the exact captured train/validation snapshots and compare bytes.""" + model = load_centroid_model(checkpoint) + return train_keyphrase_centroid(inputs, config=model.config).to_bytes() == checkpoint + + +def _document_row(item: Document) -> dict[str, object]: + return { + "id": item.id, + "title": item.title, + "abstract": item.abstract, + "topics": list(item.topics), + "keywords": list(item.keywords), + "required_experts": item.required_experts, + "metadata": dict(item.metadata), + } + + +def _expert_row(item: Expert) -> dict[str, object]: + return { + "id": item.id, + "name": item.name, + "summary": item.summary, + "topics": list(item.topics), + "keywords": list(item.keywords), + "publications": [ + { + "id": publication.id, + "title": publication.title, + "abstract": publication.abstract, + "year": publication.year, + } + for publication in item.publications + ], + "capacity": item.capacity, + "institution": item.institution, + "regions": list(item.regions), + "seniority": item.seniority, + "bids": dict(item.bids), + "metadata": dict(item.metadata), + } + + +def _affinity_bytes(scores: tuple[Affinity, ...]) -> bytes: + stream = io.StringIO(newline="") + writer = csv.writer(stream, lineterminator="\n") + writer.writerow(("document_id", "expert_id", "score")) + for item in scores: + writer.writerow((item.document_id, item.expert_id, f"{item.score:.6f}")) + return stream.getvalue().encode("utf-8", "strict") + + +def write_keyphrase_centroid_run( + inputs: CentroidInputs, + model: CentroidModel, + directory: str | Path, +) -> dict[str, object]: + """Replay and publish match-ready holdout artifacts without replacing a target.""" + destination = Path(directory) + if destination.exists() or destination.is_symlink(): + raise DataValidationError("centroid destination already exists") + if not verify_keyphrase_centroid(inputs, model.to_bytes()): + raise DataValidationError("centroid model does not replay from exact sources") + for name, path in inputs.paths.items(): + if _read(path, len(inputs.raw[name]), name) != inputs.raw[name]: + raise DataValidationError(f"centroid source changed after loading: {name}") + # Published normalized rows must still describe the captured source bytes, + # even if a caller constructed a replacement CentroidInputs object. + if ( + inputs.documents + != load_documents_text(_decode(inputs.raw["documents"], "documents"), max_records=10_000) + or inputs.experts + != load_experts_text(_decode(inputs.raw["experts"], "experts"), max_records=10_000) + or inputs.conflicts + != ( + load_conflicts_text(_decode(inputs.raw["conflicts"], "conflicts"), max_records=10_000) + if "conflicts" in inputs.raw + else () + ) + ): + raise DataValidationError("centroid normalized input objects changed after loading") + scores = score_keyphrase_centroid(inputs, model) + holdout = set(inputs.holdout_ids) + artifacts = { + "model.json": model.to_bytes(), + "documents.json": _json( + [ + _document_row(item) + for item in sorted(inputs.documents, key=lambda value: value.id) + if item.id in holdout + ] + ), + "experts.json": _json( + [_expert_row(item) for item in sorted(inputs.experts, key=lambda value: value.id)] + ), + "conflicts.json": _json( + [ + { + "document_id": item.document_id, + "expert_id": item.expert_id, + "reason": item.reason, + } + for item in sorted( + inputs.conflicts, key=lambda value: (value.document_id, value.expert_id) + ) + if item.document_id in holdout + ] + ), + "affinities.csv": _affinity_bytes(scores), + } + if sum(map(len, artifacts.values())) > 32 * 1024 * 1024: + raise DataValidationError("centroid run exceeds output byte limit") + manifest: dict[str, object] = { + "schema_version": 1, + "protocol": PROTOCOL, + "input_sha256": inputs.hashes, + "files": { + name: {"bytes": len(raw), "sha256": _sha(raw)} for name, raw in artifacts.items() + }, + "counts": { + "train_triplets": len(inputs.train), + "validation_triplets": len(inputs.validation), + "holdout_documents": len(holdout), + "reviewers": len(inputs.experts), + "affinities": len(scores), + }, + "selected_epoch": model.best_epoch, + "validation_map": model.validation_map, + } + manifest_raw = _json(manifest) + if len(manifest_raw) > 64 * 1024: + raise DataValidationError("centroid manifest exceeds output byte limit") + destination.parent.mkdir(parents=True, exist_ok=True) + with tempfile.TemporaryDirectory( + prefix=f".{destination.name}-", dir=destination.parent + ) as stage: + staging = Path(stage) + for name, raw in artifacts.items(): + (staging / name).write_bytes(raw) + (staging / "manifest.json").write_bytes(manifest_raw) + _install_directory_no_replace(staging, destination) + return manifest diff --git a/tests/test_keyphrase_centroid.py b/tests/test_keyphrase_centroid.py new file mode 100644 index 0000000..6892925 --- /dev/null +++ b/tests/test_keyphrase_centroid.py @@ -0,0 +1,456 @@ +from __future__ import annotations + +import hashlib +import json +import math +from concurrent.futures import ThreadPoolExecutor +from dataclasses import replace +from pathlib import Path + +import pytest + +import peermatchlab.keyphrase_centroid as centroid +from peermatchlab.affinity import load_affinities_csv +from peermatchlab.cli import main +from peermatchlab.io import load_json_text +from peermatchlab.keyphrase_centroid import ( + CentroidConfig, + _update, + _validation_map, + load_centroid_inputs, + load_centroid_model, + score_keyphrase_centroid, + train_keyphrase_centroid, + verify_keyphrase_centroid, + write_keyphrase_centroid_run, +) +from peermatchlab.keyphrases import write_keyphrase_run +from peermatchlab.models import DataValidationError +from peermatchlab.pipeline import run_affinity_matching + +EXAMPLE = Path(__file__).resolve().parents[1] / "examples" / "centroid" + + +def _fixture(tmp_path: Path, *, train: bytes | None = None, validation: bytes | None = None): + tmp_path.mkdir(parents=True, exist_ok=True) + documents = tmp_path / "documents.json" + experts = tmp_path / "experts.json" + train_path = tmp_path / "train.jsonl" + validation_path = tmp_path / "validation.jsonl" + documents.write_bytes((EXAMPLE / "documents.json").read_bytes()) + experts.write_bytes((EXAMPLE / "experts.json").read_bytes()) + train_path.write_bytes(train if train is not None else (EXAMPLE / "train.jsonl").read_bytes()) + validation_path.write_bytes( + validation if validation is not None else (EXAMPLE / "validation.jsonl").read_bytes() + ) + keyphrases = tmp_path / "keyphrases" + write_keyphrase_run( + keyphrases, documents_source=documents.read_bytes(), experts_source=experts.read_bytes() + ) + return keyphrases, documents, experts, train_path, validation_path + + +def _inputs(paths: tuple[Path, Path, Path, Path, Path], *, conflicts: Path | None = None): + return load_centroid_inputs(*paths, conflicts_path=conflicts) + + +def _rewrite_phrases(paths: tuple[Path, Path, Path, Path, Path], change: str) -> None: + records_path = paths[0] / "keyphrases.jsonl" + manifest_path = paths[0] / "manifest.json" + rows = [json.loads(line) for line in records_path.read_text(encoding="utf-8").splitlines()] + selected = next(row for row in rows if row["kind"] == "profile") + if change == "missing_field": + selected.pop("token_count") + elif change == "unknown_kind": + selected["kind"] = "unknown" + elif change == "wrong_evidence": + selected["evidence_id"] = "not-owner" + elif change == "bad_count": + selected["token_count"] = True + elif change == "bad_list": + selected["keyphrases"] = "terms" + elif change == "bad_phrase": + selected["keyphrases"][0].pop("score") + elif change == "bad_term": + selected["keyphrases"][0]["term"] = "Capital" + elif change == "bad_score": + selected["keyphrases"][0]["score"] = -1 + elif change == "empty_profile": + selected["keyphrases"] = [] + elif change == "missing_profile": + rows.remove(selected) + elif change == "duplicate_profile": + rows.append(dict(selected)) + elif change == "too_many_terms": + selected["keyphrases"] = [{"term": f"term{index}", "score": 0.01} for index in range(257)] + else: + raise AssertionError(change) + raw = b"".join( + json.dumps(row, sort_keys=True, separators=(",", ":")).encode() + b"\n" for row in rows + ) + records_path.write_bytes(raw) + manifest = json.loads(manifest_path.read_text(encoding="utf-8")) + manifest["records"] = len(rows) + manifest["keyphrases_bytes"] = len(raw) + manifest["keyphrases_sha256"] = hashlib.sha256(raw).hexdigest() + manifest_path.write_text(json.dumps(manifest), encoding="utf-8") + + +def test_independent_one_step_pairwise_bce_gradient_oracle() -> None: + # q=(1,0), p=(1,0), n=(0,1), margin=1. For BCEWithLogits(-margin), + # d loss/d margin = -1/(1+e), so each coordinate moves by 0.1/(1+e). + weights = [[1.0, 0.0], [1.0, 0.0], [0.0, 1.0]] + loss = _update(weights, (0,), (1,), (2,), dimensions=2, learning_rate=0.1, l2=0.0) + delta = 0.1 / (1.0 + math.e) + assert loss == pytest.approx(math.log1p(math.exp(-1.0))) + assert weights[0] == pytest.approx([1 + delta, -delta]) + assert weights[1] == pytest.approx([1 + delta, 0.0]) + assert weights[2] == pytest.approx([-delta, 1.0]) + + +def test_independent_average_precision_tie_oracle() -> None: + from peermatchlab.keyphrase_centroid import CentroidTriplet + + weights = [[1.0, 0.0], [0.0, 1.0], [1.0, 0.0]] + rows = ( + CentroidTriplet("a", "positive", "negative"), + CentroidTriplet("b", "positive", "negative"), + ) + # a: positive first -> AP 1; b: negative first -> AP 1/2. + assert _validation_map( + rows, {"a": (0,), "b": (1,)}, {"positive": (2,), "negative": (1,)}, weights, 2 + ) == pytest.approx(0.75) + + +def test_exact_source_training_checkpoint_and_match_pipeline(tmp_path: Path) -> None: + paths = _fixture(tmp_path) + inputs = _inputs(paths) + config = CentroidConfig(dimensions=2, epochs=3, seed=7) + model = train_keyphrase_centroid(inputs, config=config) + assert model.best_epoch == 1 and model.validation_map == 1.0 + assert len(model.history) == 3 + assert model.to_bytes() == train_keyphrase_centroid(inputs, config=config).to_bytes() + assert load_centroid_model(model.to_bytes()) == model + assert verify_keyphrase_centroid(inputs, model.to_bytes()) + scores = score_keyphrase_centroid(inputs, model) + assert [(item.document_id, item.expert_id) for item in scores] == [ + ("d-hold", "e-graph"), + ("d-hold", "e-math"), + ] + assert scores[1].score > scores[0].score + run = run_affinity_matching( + tuple(item for item in inputs.documents if item.id == "d-hold"), inputs.experts, scores + ) + assert run.plan.assignments[0].expert_id == "e-math" + output = tmp_path / "run" + manifest = write_keyphrase_centroid_run(inputs, model, output) + assert manifest["counts"]["affinities"] == 2 + assert load_affinities_csv(output / "affinities.csv") == scores + assert [item["id"] for item in load_json_text((output / "documents.json").read_text())] == [ + "d-hold" + ] + assert load_centroid_model((output / "model.json").read_bytes()) == model + for name, record in manifest["files"].items(): + raw = (output / name).read_bytes() + assert record == {"bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest()} + with pytest.raises(DataValidationError, match="already exists"): + write_keyphrase_centroid_run(inputs, model, output) + + +def test_loaded_source_snapshot_and_model_provenance_are_immutable(tmp_path: Path) -> None: + inputs = _inputs(_fixture(tmp_path)) + original_hashes = inputs.hashes + with pytest.raises(TypeError): + inputs.submissions["d-train"] = ("forged",) # type: ignore[index] + with pytest.raises(TypeError): + inputs.raw["train"] = b"forged" # type: ignore[index] + assert inputs.hashes == original_hashes + model = train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1)) + with pytest.raises(TypeError): + model.source_sha256["train"] = "0" * 64 # type: ignore[index] + assert verify_keyphrase_centroid(inputs, model.to_bytes()) + + +def test_writer_rejects_replaced_normalized_rows_not_in_source_bytes(tmp_path: Path) -> None: + inputs = _inputs(_fixture(tmp_path)) + model = train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1)) + forged = replace( + inputs, + documents=(replace(inputs.documents[0], title="Changed title"), *inputs.documents[1:]), + ) + with pytest.raises(DataValidationError, match="normalized input objects changed"): + write_keyphrase_centroid_run(forged, model, tmp_path / "forged-run") + assert not (tmp_path / "forged-run").exists() + + +def test_extreme_untrusted_numeric_fields_fail_as_validation_errors(tmp_path: Path) -> None: + huge = 10**1000 + with pytest.raises(DataValidationError, match="learning_rate"): + CentroidConfig(learning_rate=huge) + with pytest.raises(DataValidationError, match="l2"): + CentroidConfig(l2=huge) + model = train_keyphrase_centroid( + _inputs(_fixture(tmp_path)), config=CentroidConfig(dimensions=2, epochs=1) + ) + for mutation in ( + {"vectors": ((huge, *model.vectors[0][1:]), *model.vectors[1:])}, + {"validation_map": huge}, + {"history": ((1, huge, 1.0),)}, + {"history": ((1, 0.1, huge),)}, + ): + with pytest.raises(DataValidationError, match="invalid dimensions"): + replace(model, **mutation) + + +def test_validation_labels_do_not_enter_weight_updates(tmp_path: Path) -> None: + baseline = _inputs(_fixture(tmp_path / "baseline")) + reversed_validation = ( + (EXAMPLE / "validation.jsonl") + .read_bytes() + .replace( + b'"positive_expert_id":"e-math","negative_expert_id":"e-graph"', + b'"positive_expert_id":"e-graph","negative_expert_id":"e-math"', + ) + ) + changed = _inputs(_fixture(tmp_path / "changed", validation=reversed_validation)) + config = CentroidConfig(dimensions=2, epochs=1, seed=11) + a = train_keyphrase_centroid(baseline, config=config) + b = train_keyphrase_centroid(changed, config=config) + assert a.vocabulary == b.vocabulary and a.vectors == b.vectors + assert a.history[0][1] == b.history[0][1] + assert a.validation_map != b.validation_map + assert a.source_sha256["validation"] != b.source_sha256["validation"] + + +@pytest.mark.parametrize( + "train,validation,reason", + [ + ( + b'{"document_id":"d-train","positive_expert_id":"e-math","negative_expert_id":"e-math"}\n', + None, + "distinct", + ), + ((EXAMPLE / "train.jsonl").read_bytes() * 2, None, "repeats"), + ( + (EXAMPLE / "train.jsonl").read_bytes() + + b'{"document_id":"d-train","positive_expert_id":"e-graph",' + b'"negative_expert_id":"e-math"}\n', + None, + "contradictory", + ), + (None, (EXAMPLE / "train.jsonl").read_bytes(), "overlap"), + ( + b'{"document_id":"missing","positive_expert_id":"e-math","negative_expert_id":"e-graph"}\n', + None, + "missing", + ), + ( + b'{"document_id":"d-train","positive_expert_id":"missing","negative_expert_id":"e-graph"}\n', + None, + "missing", + ), + (b"{}\n", None, "schema"), + (b'{"document_id":"x","document_id":"y"}\n', None, "invalid train"), + (b"\xff", None, "UTF-8"), + ], +) +def test_invalid_triplets_fail_closed( + tmp_path: Path, train: bytes | None, validation: bytes | None, reason: str +) -> None: + with pytest.raises(DataValidationError, match=reason): + _inputs(_fixture(tmp_path, train=train, validation=validation)) + + +def test_hash_conflict_empty_evidence_and_resource_guards(tmp_path: Path) -> None: + paths = _fixture(tmp_path) + paths[1].write_bytes(paths[1].read_bytes() + b" ") + with pytest.raises(DataValidationError, match="hashes"): + _inputs(paths) + paths[1].write_bytes((EXAMPLE / "documents.json").read_bytes()) + conflict = tmp_path / "conflicts.json" + conflict.write_text('[{"document_id":"d-train","expert_id":"e-math"}]', encoding="utf-8") + with pytest.raises(DataValidationError, match="positive"): + _inputs(paths, conflicts=conflict) + inputs = _inputs(paths) + with pytest.raises(DataValidationError, match="max_work"): + train_keyphrase_centroid(inputs, config=CentroidConfig(max_work=1)) + for kwargs in ({"dimensions": 0}, {"epochs": 51}, {"learning_rate": 0}, {"seed": -1}): + with pytest.raises(DataValidationError): + CentroidConfig(**kwargs) + model = train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1)) + tampered = bytearray(model.to_bytes()) + tampered[-10] = ord("f") if tampered[-10] != ord("f") else ord("e") + with pytest.raises(DataValidationError): + load_centroid_model(bytes(tampered)) + with pytest.raises(DataValidationError, match="hashes"): + changed = ( + (EXAMPLE / "validation.jsonl") + .read_bytes() + .replace( + b'"positive_expert_id":"e-math","negative_expert_id":"e-graph"', + b'"positive_expert_id":"e-graph","negative_expert_id":"e-math"', + ) + ) + score_keyphrase_centroid(_inputs(_fixture(tmp_path / "other", validation=changed)), model) + + +@pytest.mark.parametrize( + "change,reason", + [ + ("missing_field", "schema"), + ("unknown_kind", "identity"), + ("wrong_evidence", "inconsistent"), + ("bad_count", "token_count"), + ("bad_list", "list"), + ("bad_phrase", "term schema"), + ("bad_term", "term or score"), + ("bad_score", "term or score"), + ("empty_profile", "missing"), + ("missing_profile", "IDs do not match"), + ("duplicate_profile", "duplicate"), + ("too_many_terms", "per-entity"), + ], +) +def test_forged_but_hash_consistent_keyphrase_records_rejected( + tmp_path: Path, change: str, reason: str +) -> None: + paths = _fixture(tmp_path) + _rewrite_phrases(paths, change) + with pytest.raises(DataValidationError, match=reason): + _inputs(paths) + + +def test_manifest_labels_partition_and_source_limits(tmp_path: Path) -> None: + paths = _fixture(tmp_path) + manifest = paths[0] / "manifest.json" + original = manifest.read_bytes() + value = json.loads(original) + value["schema_version"] = 2 + manifest.write_text(json.dumps(value), encoding="utf-8") + with pytest.raises(DataValidationError, match="unsupported"): + _inputs(paths) + manifest.write_bytes(original) + value = json.loads(original) + value["records"] = 500 + manifest.write_text(json.dumps(value), encoding="utf-8") + with pytest.raises(DataValidationError, match="record count mismatch"): + _inputs(paths) + manifest.write_bytes(original) + paths[3].write_bytes(b"") + with pytest.raises(DataValidationError, match="empty"): + _inputs(paths) + paths[3].write_bytes(b"0\n") + with pytest.raises(DataValidationError, match="object"): + _inputs(paths) + paths[3].write_bytes(b" " * (2 * 1024 * 1024 + 1)) + with pytest.raises(DataValidationError, match="exceeds"): + _inputs(paths) + paths[3].write_bytes((EXAMPLE / "train.jsonl").read_bytes()) + with pytest.raises(DataValidationError, match="distinct"): + load_centroid_inputs(paths[0], paths[1], paths[2], paths[3], paths[3]) + conflicts = tmp_path / "conflicts.json" + conflicts.write_text('[{"document_id":"missing","expert_id":"e-math"}]', encoding="utf-8") + with pytest.raises(DataValidationError, match="unknown"): + _inputs(paths, conflicts=conflicts) + conflicts.write_text( + '[{"document_id":"d-hold","expert_id":"e-math"},' + '{"document_id":"d-hold","expert_id":"e-math"}]', + encoding="utf-8", + ) + with pytest.raises(DataValidationError, match="duplicate conflict"): + _inputs(paths, conflicts=conflicts) + + +def test_checkpoint_shape_and_publication_source_drift( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + paths = _fixture(tmp_path) + inputs = _inputs(paths) + model = train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1)) + envelope = json.loads(model.to_bytes()) + envelope["payload"]["vectors"][0] = [True, False] + envelope["sha256"] = hashlib.sha256(centroid._json(envelope["payload"])).hexdigest() + with pytest.raises(DataValidationError, match="fields are invalid"): + load_centroid_model(centroid._json(envelope)) + with pytest.raises(DataValidationError, match="byte limit"): + load_centroid_model(b" " * (8 * 1024 * 1024 + 1)) + with pytest.raises(DataValidationError, match="source changed"): + paths[3].write_bytes(paths[3].read_bytes().replace(b"d-train", b"D-train")) + write_keyphrase_centroid_run(inputs, model, tmp_path / "not-published") + assert not (tmp_path / "not-published").exists() + monkeypatch.setattr(centroid, "_MAX_TERMS", 1) + with pytest.raises(DataValidationError, match="vocabulary"): + train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1)) + + +def test_cli_create_only_and_complete_match_smoke( + tmp_path: Path, capsys: pytest.CaptureFixture[str] +) -> None: + paths = _fixture(tmp_path) + directory = tmp_path / "cli-run" + args = [ + "expertise-centroid", + "--keyphrases", + str(paths[0]), + "--documents", + str(paths[1]), + "--experts", + str(paths[2]), + "--train", + str(paths[3]), + "--validation", + str(paths[4]), + "--dimensions", + "2", + "--epochs", + "2", + "--directory", + str(directory), + ] + assert main(args) == 0 + assert "generated 2 holdout affinities" in capsys.readouterr().out + assert main(args) == 2 + assert "already exists" in capsys.readouterr().err + assert ( + main( + [ + "match-affinity", + "--documents", + str(directory / "documents.json"), + "--experts", + str(directory / "experts.json"), + "--conflicts", + str(directory / "conflicts.json"), + "--affinities", + str(directory / "affinities.csv"), + "--config", + str(EXAMPLE.parent / "expertise" / "match-config.json"), + "--output", + str(tmp_path / "assignment.json"), + ] + ) + == 0 + ) + assert ( + json.loads((tmp_path / "assignment.json").read_text())["assignments"][0]["expert_id"] + == "e-math" + ) + + +def test_concurrent_create_only_run_has_one_winner(tmp_path: Path) -> None: + inputs = _inputs(_fixture(tmp_path)) + model = train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1)) + output = tmp_path / "race" + with ThreadPoolExecutor(max_workers=2) as pool: + futures = [ + pool.submit(write_keyphrase_centroid_run, inputs, model, output) for _ in range(2) + ] + outcomes = [] + for future in futures: + try: + future.result() + outcomes.append("success") + except (DataValidationError, OSError): + outcomes.append("exists") + assert sorted(outcomes) == ["exists", "success"] + assert load_centroid_model((output / "model.json").read_bytes()) == model diff --git a/uv.lock b/uv.lock index f0c3b02..94ed79f 100644 --- a/uv.lock +++ b/uv.lock @@ -478,7 +478,7 @@ wheels = [ [[package]] name = "peermatchlab" -version = "0.15.0" +version = "0.16.0" source = { editable = "." } [package.optional-dependencies] From 5458e1e1cca47cfc624b6112188c050050b5476b Mon Sep 17 00:00:00 2001 From: appleweiping Date: Sun, 20 Sep 2026 03:02:42 -0500 Subject: [PATCH 2/2] Bind centroid labels and features to captured sources Signed-off-by: appleweiping --- docs/trainable-centroid.md | 6 ++- src/peermatchlab/keyphrase_centroid.py | 59 ++++++++++++++++++++------ tests/test_keyphrase_centroid.py | 37 +++++++++++++++- 3 files changed, 88 insertions(+), 14 deletions(-) diff --git a/docs/trainable-centroid.md b/docs/trainable-centroid.md index c3ec3eb..272b26a 100644 --- a/docs/trainable-centroid.md +++ b/docs/trainable-centroid.md @@ -49,6 +49,8 @@ command overwrites existing files. The run contains a versioned `model.json`, checkpoint stores a learned vocabulary and weights, so a real venue must handle it as private derived data; the manifest does not include raw text or labels. Preserve the original inputs to replay or audit the declared hashes. +These hashes detect byte changes against retained local sources; they are not +signatures or proof that caller-authored labels or evidence are trustworthy. ## Data separation and scoring @@ -100,7 +102,9 @@ hashes, or a changed input path before publication are errors. `load_centroid_model` verifies the checkpoint's canonical payload digest and shape; `verify_keyphrase_centroid` refits from the exact captured inputs and -compares checkpoint bytes. The run writer performs that replay and rechecks +compares checkpoint bytes. Training and scoring re-read the source paths and +reject replaced in-memory labels or feature mappings that disagree with the +captured bytes. The run writer performs that replay and rechecks source paths before publishing a staged directory with no replacement. An independent hand calculation in the tests checks one pairwise BCE gradient step; a separate two-submission ranking oracle checks MAP. Real-world use diff --git a/src/peermatchlab/keyphrase_centroid.py b/src/peermatchlab/keyphrase_centroid.py index 46be3d5..eef1527 100644 --- a/src/peermatchlab/keyphrase_centroid.py +++ b/src/peermatchlab/keyphrase_centroid.py @@ -80,6 +80,14 @@ def _decode(raw: bytes, label: str) -> str: raise DataValidationError(f"{label} must be strict UTF-8") from error +def _source_limit(name: str) -> int: + if name == "keyphrase-manifest.json": + return 64 * 1024 + if name in {"train", "validation", "conflicts"}: + return _MAX_LABEL_SOURCE + return _MAX_SOURCE + + def _json_lines(raw: bytes, label: str, maximum: int) -> tuple[dict[str, Any], ...]: rows: list[dict[str, Any]] = [] for number, line in enumerate(_decode(raw, label).splitlines(), 1): @@ -302,18 +310,7 @@ def load_centroid_inputs( paths["conflicts"] = Path(conflicts_path) if len({path.resolve(strict=False) for path in paths.values()}) != len(paths): raise DataValidationError("centroid input paths must be distinct") - raw = { - name: _read( - path, - 64 * 1024 - if name == "keyphrase-manifest.json" - else _MAX_LABEL_SOURCE - if name in {"train", "validation", "conflicts"} - else _MAX_SOURCE, - name, - ) - for name, path in paths.items() - } + raw = {name: _read(path, _source_limit(name), name) for name, path in paths.items()} try: manifest = load_json_text(_decode(raw["keyphrase-manifest.json"], "keyphrase manifest")) except (ValueError, RecursionError) as error: @@ -401,6 +398,42 @@ def load_centroid_inputs( return result +def _assert_captured_inputs(inputs: CentroidInputs) -> None: + """Reject replaced fields that no longer describe the exact source snapshot.""" + required = { + "keyphrases.jsonl", + "keyphrase-manifest.json", + "documents", + "experts", + "train", + "validation", + } + names = required | ({"conflicts"} if "conflicts" in inputs.paths else set()) + if ( + set(inputs.paths) != names + or set(inputs.raw) != names + or any(not isinstance(path, Path) for path in inputs.paths.values()) + or any( + type(raw) is not bytes or len(raw) > _source_limit(name) + for name, raw in inputs.raw.items() + ) + ): + raise DataValidationError("centroid input snapshot paths or bytes are malformed") + for name, path in inputs.paths.items(): + if _read(path, len(inputs.raw[name]), name) != inputs.raw[name]: + raise DataValidationError(f"centroid source changed after loading: {name}") + fresh = load_centroid_inputs( + inputs.paths["keyphrases.jsonl"].parent, + inputs.paths["documents"], + inputs.paths["experts"], + inputs.paths["train"], + inputs.paths["validation"], + conflicts_path=inputs.paths.get("conflicts"), + ) + if fresh != inputs: + raise DataValidationError("centroid input snapshot differs from exact source bytes") + + def _centroid( indices: tuple[int, ...], weights: list[list[float]] | tuple[tuple[float, ...], ...], @@ -663,6 +696,7 @@ def train_keyphrase_centroid( """Train on train triplets only; choose first highest-MAP validation epoch.""" if not isinstance(inputs, CentroidInputs): raise DataValidationError("centroid training requires validated inputs") + _assert_captured_inputs(inputs) chosen = config or CentroidConfig() train_docs = {row.document_id for row in inputs.train} train_reviewers = { @@ -766,6 +800,7 @@ def score_keyphrase_centroid(inputs: CentroidInputs, model: CentroidModel) -> tu """Score only unseen holdout submissions for the existing assignment API.""" if not isinstance(inputs, CentroidInputs) or not isinstance(model, CentroidModel): raise DataValidationError("centroid scoring requires validated inputs and model") + _assert_captured_inputs(inputs) if model.source_sha256 != inputs.hashes: raise DataValidationError("centroid checkpoint source hashes do not match current inputs") if len(inputs.holdout_ids) * len(inputs.experts) > _MAX_PAIRS: diff --git a/tests/test_keyphrase_centroid.py b/tests/test_keyphrase_centroid.py index 6892925..5b55d70 100644 --- a/tests/test_keyphrase_centroid.py +++ b/tests/test_keyphrase_centroid.py @@ -178,11 +178,44 @@ def test_writer_rejects_replaced_normalized_rows_not_in_source_bytes(tmp_path: P inputs, documents=(replace(inputs.documents[0], title="Changed title"), *inputs.documents[1:]), ) - with pytest.raises(DataValidationError, match="normalized input objects changed"): + with pytest.raises(DataValidationError, match="snapshot differs"): write_keyphrase_centroid_run(forged, model, tmp_path / "forged-run") assert not (tmp_path / "forged-run").exists() +def test_train_and_score_reject_replaced_labels_or_features_with_old_raw( + tmp_path: Path, +) -> None: + inputs = _inputs(_fixture(tmp_path)) + model = train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1)) + changed_train = replace( + inputs, + train=( + replace( + inputs.train[0], + positive_expert_id=inputs.train[0].negative_expert_id, + negative_expert_id=inputs.train[0].positive_expert_id, + ), + ), + ) + changed_validation = replace(inputs, validation=()) + changed_submission = replace(inputs, submissions={**inputs.submissions, "d-train": ("forged",)}) + changed_reviewer = replace(inputs, reviewers={**inputs.reviewers, "e-math": ("forged",)}) + for forged in ( + changed_train, + changed_validation, + changed_submission, + changed_reviewer, + ): + with pytest.raises(DataValidationError, match="snapshot differs"): + train_keyphrase_centroid(forged, config=model.config) + with pytest.raises(DataValidationError, match="snapshot differs"): + score_keyphrase_centroid(forged, model) + with pytest.raises(DataValidationError, match="snapshot differs"): + write_keyphrase_centroid_run(forged, model, tmp_path / "forged-run") + assert not (tmp_path / "forged-run").exists() + + def test_extreme_untrusted_numeric_fields_fail_as_validation_errors(tmp_path: Path) -> None: huge = 10**1000 with pytest.raises(DataValidationError, match="learning_rate"): @@ -374,10 +407,12 @@ def test_checkpoint_shape_and_publication_source_drift( load_centroid_model(centroid._json(envelope)) with pytest.raises(DataValidationError, match="byte limit"): load_centroid_model(b" " * (8 * 1024 * 1024 + 1)) + train_original = paths[3].read_bytes() with pytest.raises(DataValidationError, match="source changed"): paths[3].write_bytes(paths[3].read_bytes().replace(b"d-train", b"D-train")) write_keyphrase_centroid_run(inputs, model, tmp_path / "not-published") assert not (tmp_path / "not-published").exists() + paths[3].write_bytes(train_original) monkeypatch.setattr(centroid, "_MAX_TERMS", 1) with pytest.raises(DataValidationError, match="vocabulary"): train_keyphrase_centroid(inputs, config=CentroidConfig(dimensions=2, epochs=1))