From 865373ca7bf5984a42f74187b107987372518eb0 Mon Sep 17 00:00:00 2001 From: appleweiping Date: Sat, 19 Sep 2026 23:54:19 -0500 Subject: [PATCH] Add bounded offline OpenReview assignment publishing plan Signed-off-by: appleweiping --- CHANGELOG.md | 10 + CITATION.cff | 2 +- README.md | 4 + docs/openreview-publish.md | 65 +++ docs/releasing.md | 4 +- examples/openreview_publish_mock.py | 68 +++ pyproject.toml | 2 +- src/peermatchlab/__init__.py | 22 +- src/peermatchlab/openreview_publish.py | 461 +++++++++++++++++++++ tests/test_openreview_publish.py | 547 +++++++++++++++++++++++++ uv.lock | 2 +- 11 files changed, 1181 insertions(+), 6 deletions(-) create mode 100644 docs/openreview-publish.md create mode 100644 examples/openreview_publish_mock.py create mode 100644 src/peermatchlab/openreview_publish.py create mode 100644 tests/test_openreview_publish.py diff --git a/CHANGELOG.md b/CHANGELOG.md index ee64967..21ae87c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,16 @@ All notable changes are documented here. The project follows semantic versioning _No changes yet._ +## 0.13.0 - 2026-09-19 + +- Added an offline-first, bounded OpenReview assignment/aggregate-score edge plan + with exact invitation, group, paper, reviewer, and digest validation. Real + writes require an explicitly injected transport and matching SHA-256 + confirmation; dry-run is the default. Per-batch read-back supports safe + reconciliation after ambiguous partial failures, with one atomic + no-overwrite local plan/receipt evidence file and synthetic mock integration + tests. No live API write adapter or venue credentials are shipped. + ## 0.12.0 - 2026-09-19 - Added an opt-in exact, bounded `maximin` strategy for small reviewer panels. diff --git a/CITATION.cff b/CITATION.cff index 8b76ff0..cf44315 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -6,7 +6,7 @@ authors: - family-names: "Yan" given-names: "Weiping" alias: "appleweiping" -version: 0.12.0 +version: 0.13.0 date-released: 2026-09-19 license: MIT repository-code: "https://github.com/appleweiping/PeerMatchLab" diff --git a/README.md b/README.md index 2e90d2c..7da3687 100644 --- a/README.md +++ b/README.md @@ -414,6 +414,10 @@ only direct group membership and submission metadata; the explicit expertise option additionally retrieves scoped profile/publication evidence. Neither mode infers conflicts, expands nested groups, or downloads attachments. +For an offline-first assignment edge plan and a synthetic in-memory write-back +replay, see [assignment edge preparation](docs/openreview-publish.md). The +default is dry-run; no live OpenReview write adapter ships with this package. + ## Python API ```python diff --git a/docs/openreview-publish.md b/docs/openreview-publish.md new file mode 100644 index 0000000..39e5a5b --- /dev/null +++ b/docs/openreview-publish.md @@ -0,0 +1,65 @@ +# Assignment edge preparation and mock write-back + +This is an offline-first protocol slice, **not a production OpenReview publisher**. +The frozen Matcher reference creates an assignment edge and a matching aggregate-score +edge for every paper/reviewer pair. PeerMatchLab prepares those two edge sets and +audits an independently produced `MatchPlan`; it does not ship an HTTP POST client, +authentication, invitation discovery, or a hosted matching service. No test or +example contacts a real venue. + +`prepare_publish_plan` accepts the exact assignment invitation, score invitation, +reviewer group, label, paper IDs, and reviewer IDs in a `PublishScope`. The IDs must +exactly match the audited domain inputs. The plan must have complete demand, no +conflicts, capacity excess, duplicate pair, or other hard-audit violation. Input +scores are copied verbatim to both invitation-specific edges; the canonical JSON +of all scoped IDs and edge content determines the SHA-256 plan digest. The maximum +is 1,000 assignments and 2,000 edges. This preparation makes no network calls. + +`publish_assignment_plan` is dry-run by default, even if a transport was supplied. +An actual write requires **all** of `publish=True`, a caller-supplied +`AssignmentWriteTransport`, and `confirm_sha256` equal to the immutable plan digest. +The transport must fetch exact invitation IDs, group membership, and paper IDs, +and must return the complete set of edges for the specified two invitations, +label, and paper heads. Both returned context and existing edge identity/content +are checked before a write. Any unexpected or divergent edge aborts; this code +never deletes or overwrites an edge **provided the injected adapter honors its +required atomic create-only contract**. Each edge insertion must fail on an +existing identity; a plain OpenReview bulk upsert cannot safely implement this +transport because another actor may insert an edge after the preflight read. +Batches contain at most 100 edges. + +After every batch, the context and complete edge set are read back. An ambiguous +partial transport failure, unavailable read-back, or not-yet-visible edge returns +an `incomplete` receipt, not a blind automatic retry. Repeating the call re-reads +the remote set and sends only missing edges; existing exact edges produce a +zero-write `complete` receipt. A concurrent actor that changes a label-scoped +edge or venue context causes fail-closed behavior when the conditional-insert +contract is upheld. Operators must inspect +`confirmed_edges`, `failure`, and the authoritative venue state before retrying. +`write_publish_artifact` stores both objects in one canonical JSON file. It +writes a temporary file beside the requested path and hard-links it into place, +so an existing or concurrently created destination is never replaced. The +temporary name is removed even when installation fails. Paper IDs are limited +to 1,000, reviewer IDs to 10,000, and conflicts to 100,000; generator inputs +are read only up to one record past these ceilings before rejection. + +Try the purely in-memory synthetic replay: + +```bash +uv run python examples/openreview_publish_mock.py +``` + +It prints one digest, then `dry_run`, `complete`, and `0` attempted batches +on an idempotent re-run. `tests/test_openreview_publish.py` also simulates a +partial POST, stale read-back, changed context, duplicate/foreign edges, and +read-back failure. There are no credentials, real invitation IDs, or network +operations in either sample. + +This boundary intentionally leaves OpenReview API v1/v2 edge payload shape, +invitation authorization, pagination of existing edges, eventual-consistency +timing, atomic conditional-insert support, and production conflict policy to a +future separately reviewed adapter. If a real endpoint cannot provide atomic +create-only semantics, the write path must remain disabled; preflight alone +is insufficient. +The caller must not treat a mock-transport `complete` receipt as proof that a +real venue was updated. diff --git a/docs/releasing.md b/docs/releasing.md index 68164e7..7a4a29d 100644 --- a/docs/releasing.md +++ b/docs/releasing.md @@ -1,6 +1,6 @@ # Release process -A push of a semantic-version tag such as `v0.12.0` invokes the release +A push of a semantic-version tag such as `v0.13.0` invokes the release workflow. The workflow refuses a tag that differs from `project.version`, installs the committed `uv.lock`, runs the static and coverage-gated suite (including a separate ≥90% covered-branches/total-branches check), @@ -17,7 +17,7 @@ Verify a downloaded file with: ```bash sha256sum --check SHA256SUMS -gh attestation verify peermatchlab-0.12.0-py3-none-any.whl \ +gh attestation verify peermatchlab-0.13.0-py3-none-any.whl \ --repo appleweiping/PeerMatchLab ``` diff --git a/examples/openreview_publish_mock.py b/examples/openreview_publish_mock.py new file mode 100644 index 0000000..5a4ab79 --- /dev/null +++ b/examples/openreview_publish_mock.py @@ -0,0 +1,68 @@ +"""Synthetic assignment-publish replay; no HTTP client or venue credentials.""" + +from peermatchlab import ( + Assignment, + AssignmentEdge, + Document, + Expert, + MatchPlan, + PublishScope, + RemoteContext, + prepare_publish_plan, + publish_assignment_plan, +) + + +class MemoryTransport: + def __init__(self) -> None: + self.edges: dict[tuple[str, str, str, str], AssignmentEdge] = {} + + def inspect_context(self, scope: PublishScope) -> RemoteContext: + return RemoteContext( + scope.assignment_invitation, + scope.score_invitation, + scope.reviewer_group, + scope.paper_ids, + scope.reviewer_ids, + ) + + def read_edges(self, scope: PublishScope) -> tuple[AssignmentEdge, ...]: + return tuple(self.edges.values()) + + def post_edges(self, edges: tuple[AssignmentEdge, ...]) -> None: + for edge in edges: + if edge.key in self.edges: + raise RuntimeError("synthetic conditional-insert conflict") + self.edges[edge.key] = edge + + +documents = (Document("synthetic-paper", "Synthetic retrieval paper", required_experts=1),) +experts = (Expert("~SyntheticReviewer1", "Synthetic Reviewer", capacity=1),) +plan = MatchPlan( + (Assignment("synthetic-paper", "~SyntheticReviewer1", 0.75, 1),), + {"synthetic-paper": 0}, + "synthetic", + 0.75, +) +scope = PublishScope( + "Synthetic/-/Assignment", + "Synthetic/-/Aggregate_Score", + "Synthetic/Reviewers", + "peer-matchlab-demo", + ("synthetic-paper",), + ("~SyntheticReviewer1",), +) +prepared = prepare_publish_plan(plan, scope, documents=documents, experts=experts, default_demand=1) +transport = MemoryTransport() +print(prepared.digest) +print(publish_assignment_plan(prepared, transport=transport).status) +print( + publish_assignment_plan( + prepared, transport=transport, publish=True, confirm_sha256=prepared.digest + ).status +) +print( + publish_assignment_plan( + prepared, transport=transport, publish=True, confirm_sha256=prepared.digest + ).attempted_batches +) diff --git a/pyproject.toml b/pyproject.toml index bb0df32..08d5bc5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "peermatchlab" -version = "0.12.0" +version = "0.13.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 6e04b18..3c1f2ba 100644 --- a/src/peermatchlab/__init__.py +++ b/src/peermatchlab/__init__.py @@ -75,6 +75,17 @@ fetch_reviewer_expertise, write_openreview_snapshot, ) +from peermatchlab.openreview_publish import ( + AssignmentEdge, + AssignmentWriteTransport, + PublishPlan, + PublishReceipt, + PublishScope, + RemoteContext, + prepare_publish_plan, + publish_assignment_plan, + write_publish_artifact, +) from peermatchlab.scoring import MatchScorer, ScoreWeights __all__ = [ @@ -82,8 +93,10 @@ "AffinityScorer", "Assignment", "AssignmentDiagnostics", + "AssignmentEdge", "AssignmentEngine", "AssignmentStrategy", + "AssignmentWriteTransport", "DemandDiagnostic", "Document", "EmbeddingProvider", @@ -117,6 +130,10 @@ "OpenReviewSnapshot", "OpenReviewTransportError", "Publication", + "PublishPlan", + "PublishReceipt", + "PublishScope", + "RemoteContext", "RetryPolicy", "ScoreWeights", "TermContribution", @@ -135,6 +152,8 @@ "load_reviewer_ids", "local_domain_inputs", "openreview_submissions_from_records", + "prepare_publish_plan", + "publish_assignment_plan", "rank_keyphrases", "read_keyphrase_source", "request_sha256", @@ -144,6 +163,7 @@ "write_expertise_run", "write_keyphrase_run", "write_openreview_snapshot", + "write_publish_artifact", ] -__version__ = "0.12.0" +__version__ = "0.13.0" diff --git a/src/peermatchlab/openreview_publish.py b/src/peermatchlab/openreview_publish.py new file mode 100644 index 0000000..cac2c10 --- /dev/null +++ b/src/peermatchlab/openreview_publish.py @@ -0,0 +1,461 @@ +"""Offline-first assignment edge plans and an injectable, fail-closed write boundary. + +This module intentionally does not contain an OpenReview HTTP POST implementation. +An operator must supply a transport appropriate to an authorized venue. Tests use +only an in-memory transport; importing or preparing a plan never contacts a service. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import tempfile +from collections.abc import Iterable, Mapping +from dataclasses import dataclass +from itertools import islice +from pathlib import Path +from typing import Protocol, TypeVar + +from peermatchlab.audit import audit_plan +from peermatchlab.models import Conflict, DataValidationError, Document, Expert, MatchPlan + +_MAX_ASSIGNMENTS = 1_000 +_MAX_BATCH_SIZE = 100 +_MAX_PAPERS = 1_000 +_MAX_REVIEWERS = 10_000 +_MAX_CONFLICTS = 100_000 +_Record = TypeVar("_Record") + + +def _exact_id(value: object, name: str) -> str: + if ( + not isinstance(value, str) + or not value + or value != value.strip() + or not value.isprintable() + or any(character.isspace() for character in value) + or len(value.encode("utf-8")) > 512 + ): + raise DataValidationError( + f"{name} must be an exact, printable identifier of at most 512 bytes" + ) + return value + + +def _bounded_unique(values: Iterable[str], name: str, *, maximum: int) -> tuple[str, ...]: + items = tuple(_exact_id(value, name) for value in islice(values, maximum + 1)) + if not items or len(items) > maximum or len(items) != len(set(items)): + raise DataValidationError(f"{name} must contain 1 to {maximum} unique exact identifiers") + return tuple(sorted(items)) + + +def _bounded_records(values: Iterable[_Record], name: str, *, maximum: int) -> tuple[_Record, ...]: + items = tuple(islice(values, maximum + 1)) + if len(items) > maximum: + raise DataValidationError(f"{name} exceeds maximum of {maximum} records") + return items + + +@dataclass(frozen=True, slots=True) +class PublishScope: + """Venue identifiers that are bound into one immutable write plan.""" + + assignment_invitation: str + score_invitation: str + reviewer_group: str + label: str + paper_ids: tuple[str, ...] + reviewer_ids: tuple[str, ...] + + def __post_init__(self) -> None: + for name in ("assignment_invitation", "score_invitation", "reviewer_group", "label"): + _exact_id(getattr(self, name), name) + if self.assignment_invitation == self.score_invitation: + raise DataValidationError("assignment and score invitations must differ") + object.__setattr__( + self, "paper_ids", _bounded_unique(self.paper_ids, "paper_ids", maximum=_MAX_PAPERS) + ) + object.__setattr__( + self, + "reviewer_ids", + _bounded_unique(self.reviewer_ids, "reviewer_ids", maximum=_MAX_REVIEWERS), + ) + + def as_dict(self) -> dict[str, object]: + return { + "assignment_invitation": self.assignment_invitation, + "score_invitation": self.score_invitation, + "reviewer_group": self.reviewer_group, + "label": self.label, + "paper_ids": list(self.paper_ids), + "reviewer_ids": list(self.reviewer_ids), + } + + +@dataclass(frozen=True, slots=True) +class AssignmentEdge: + """The exact content of an assignment or score edge in the scoped protocol.""" + + invitation: str + head: str + tail: str + label: str + weight: float + + def __post_init__(self) -> None: + for name in ("invitation", "head", "tail", "label"): + _exact_id(getattr(self, name), name) + if isinstance(self.weight, bool) or not isinstance(self.weight, (int, float)): + raise DataValidationError("edge weight must be a finite score in [0, 1]") + if not 0 <= self.weight <= 1: + raise DataValidationError("edge weight must be a finite score in [0, 1]") + + @property + def key(self) -> tuple[str, str, str, str]: + return (self.invitation, self.head, self.tail, self.label) + + def as_dict(self) -> dict[str, object]: + return { + "invitation": self.invitation, + "head": self.head, + "tail": self.tail, + "label": self.label, + "weight": self.weight, + } + + +@dataclass(frozen=True, slots=True) +class PublishPlan: + """Canonical two-invitation edge set and SHA-256 of its exact scoped content.""" + + scope: PublishScope + edges: tuple[AssignmentEdge, ...] + digest: str + + def as_dict(self) -> dict[str, object]: + return { + "schema": "peermatchlab-openreview-edge-plan-v1", + "scope": self.scope.as_dict(), + "edges": [edge.as_dict() for edge in self.edges], + "sha256": self.digest, + } + + +def _canonical_bytes(value: Mapping[str, object]) -> bytes: + return ( + json.dumps( + value, sort_keys=True, separators=(",", ":"), ensure_ascii=False, allow_nan=False + ) + + "\n" + ).encode("utf-8") + + +def prepare_publish_plan( + plan: MatchPlan, + scope: PublishScope, + *, + documents: Iterable[Document], + experts: Iterable[Expert], + conflicts: Iterable[Conflict] = (), + default_demand: int = 2, + require_distinct_institutions: bool = False, +) -> PublishPlan: + """Reject unsafe/incomplete plans before any external transport is consulted.""" + + document_items = _bounded_records(documents, "documents", maximum=_MAX_PAPERS) + expert_items = _bounded_records(experts, "experts", maximum=_MAX_REVIEWERS) + conflict_items = _bounded_records(conflicts, "conflicts", maximum=_MAX_CONFLICTS) + if set(scope.paper_ids) != {item.id for item in document_items}: + raise DataValidationError("paper_ids must exactly match audited documents") + if set(scope.reviewer_ids) != {item.id for item in expert_items}: + raise DataValidationError("reviewer_ids must exactly match audited experts") + if not plan.assignments or len(plan.assignments) > _MAX_ASSIGNMENTS: + raise DataValidationError(f"publish plan requires 1 to {_MAX_ASSIGNMENTS} assignments") + audit = audit_plan( + plan, + document_items, + expert_items, + conflict_items, + default_demand=default_demand, + require_distinct_institutions=require_distinct_institutions, + ) + if not audit.safe or audit.demand_coverage != 1 or any(plan.unmet.values()): + raise DataValidationError( + "only independently audited, complete assignments can be published" + ) + edges = tuple( + sorted( + ( + AssignmentEdge( + invitation, item.document_id, item.expert_id, scope.label, item.score + ) + for item in plan.assignments + for invitation in (scope.assignment_invitation, scope.score_invitation) + ), + key=lambda item: item.key, + ) + ) + if len({edge.key for edge in edges}) != len(edges): + raise DataValidationError("publish plan contains duplicate edges") + body: dict[str, object] = { + "schema": "peermatchlab-openreview-edge-plan-v1", + "scope": scope.as_dict(), + "edges": [edge.as_dict() for edge in edges], + } + return PublishPlan(scope, edges, hashlib.sha256(_canonical_bytes(body)).hexdigest()) + + +@dataclass(frozen=True, slots=True) +class RemoteContext: + """Exact invitation/group/paper evidence fetched by an injected transport.""" + + assignment_invitation: str + score_invitation: str + reviewer_group: str + paper_ids: tuple[str, ...] + reviewer_ids: tuple[str, ...] + + +class AssignmentWriteTransport(Protocol): + """Operator-provided boundary requiring atomic create-only edge writes. + + An adapter based on unconditional bulk upsert does not implement this + contract: it could replace an edge inserted between read and write. + """ + + def inspect_context(self, scope: PublishScope) -> RemoteContext: + """Read exact invitation IDs, group members, and paper IDs.""" + + def read_edges(self, scope: PublishScope) -> tuple[AssignmentEdge, ...]: + """Read *all* edges for both invitations, label, and paper IDs in scope.""" + + def post_edges(self, edges: tuple[AssignmentEdge, ...]) -> None: + """Atomically create each edge or conflict on an existing identity. + + Never overwrite an existing edge, even when its content is identical. + A batch may partially succeed before a conflict/transport error. + """ + + +@dataclass(frozen=True, slots=True) +class PublishReceipt: + """Auditable result; incomplete results are safe to reconcile and retry.""" + + plan_sha256: str + status: str + expected_edges: int + confirmed_edges: int + attempted_batches: int + failure: str | None = None + + def __post_init__(self) -> None: + if len(self.plan_sha256) != 64 or any( + char not in "0123456789abcdef" for char in self.plan_sha256 + ): + raise DataValidationError("receipt plan_sha256 must be a lowercase SHA-256 digest") + if self.status not in {"dry_run", "incomplete", "complete"}: + raise DataValidationError("receipt status is unknown") + for name in ("expected_edges", "confirmed_edges", "attempted_batches"): + value = getattr(self, name) + if isinstance(value, bool) or not isinstance(value, int) or value < 0: + raise DataValidationError(f"receipt {name} must be a non-negative integer") + if self.expected_edges > 2 * _MAX_ASSIGNMENTS or self.confirmed_edges > self.expected_edges: + raise DataValidationError("receipt edge counts exceed the bounded plan") + if self.status == "complete" and self.confirmed_edges != self.expected_edges: + raise DataValidationError("complete receipt must confirm every edge") + if self.status == "dry_run" and (self.confirmed_edges or self.attempted_batches): + raise DataValidationError("dry-run receipt cannot claim remote writes or confirmation") + if (self.status == "incomplete") != (self.failure is not None): + raise DataValidationError("only incomplete receipts carry a failure code") + if self.failure is not None and self.failure not in { + "unconfirmed_batch", + "readback_unavailable", + }: + raise DataValidationError("receipt failure code is unknown") + + def as_dict(self) -> dict[str, object]: + return { + "schema": "peermatchlab-openreview-publish-receipt-v1", + "plan_sha256": self.plan_sha256, + "status": self.status, + "expected_edges": self.expected_edges, + "confirmed_edges": self.confirmed_edges, + "attempted_batches": self.attempted_batches, + "failure": self.failure, + } + + +def _verify_plan(plan: PublishPlan) -> None: + body = plan.as_dict() + body.pop("sha256") + if hashlib.sha256(_canonical_bytes(body)).hexdigest() != plan.digest: + raise DataValidationError("publish plan digest does not match its content") + if not plan.edges or len(plan.edges) > 2 * _MAX_ASSIGNMENTS or len(plan.edges) % 2: + raise DataValidationError("publish plan has an invalid edge count") + invitations_by_pair: dict[tuple[str, str], set[str]] = {} + weights: dict[tuple[str, str], float] = {} + for edge in plan.edges: + if ( + edge.invitation not in {plan.scope.assignment_invitation, plan.scope.score_invitation} + or edge.head not in plan.scope.paper_ids + or edge.tail not in plan.scope.reviewer_ids + or edge.label != plan.scope.label + ): + raise DataValidationError("publish plan contains an edge outside its exact scope") + pair = (edge.head, edge.tail) + invitations_by_pair.setdefault(pair, set()).add(edge.invitation) + if pair in weights and weights[pair] != edge.weight: + raise DataValidationError("assignment and score edge weights must agree") + weights[pair] = edge.weight + if any( + invitations != {plan.scope.assignment_invitation, plan.scope.score_invitation} + for invitations in invitations_by_pair.values() + ) or len({edge.key for edge in plan.edges}) != len(plan.edges): + raise DataValidationError( + "publish plan must contain exactly one edge per invitation and pair" + ) + + +def _remote_edges(plan: PublishPlan, transport: AssignmentWriteTransport) -> set[AssignmentEdge]: + rows = transport.read_edges(plan.scope) + if not isinstance(rows, tuple) or len(rows) > 2 * _MAX_ASSIGNMENTS: + raise DataValidationError("remote edge response exceeds the bounded exact scope") + if any(not isinstance(row, AssignmentEdge) for row in rows): + raise DataValidationError("remote edge response contains a malformed edge") + keys = [row.key for row in rows] + if len(keys) != len(set(keys)): + raise DataValidationError("remote edge response contains duplicate identities") + expected = set(plan.edges) + if any(row not in expected for row in rows): + raise DataValidationError("remote edge differs from or exceeds the immutable plan") + return set(rows) + + +def _verify_context(scope: PublishScope, context: RemoteContext) -> None: + if not isinstance(context, RemoteContext): + raise DataValidationError("remote context must contain exact venue evidence") + papers = _bounded_unique(context.paper_ids, "remote paper_ids", maximum=1_000) + reviewers = _bounded_unique(context.reviewer_ids, "remote reviewer_ids", maximum=10_000) + if ( + context.assignment_invitation != scope.assignment_invitation + or context.score_invitation != scope.score_invitation + or context.reviewer_group != scope.reviewer_group + or papers != scope.paper_ids + or reviewers != scope.reviewer_ids + ): + raise DataValidationError("remote invitations, group membership, or papers changed") + + +def publish_assignment_plan( + plan: PublishPlan, + *, + transport: AssignmentWriteTransport | None = None, + publish: bool = False, + confirm_sha256: str | None = None, + batch_size: int = 50, +) -> PublishReceipt: + """Dry-run by default; explicit writes reconcile every batch, never overwrite. + + An ambiguous POST returns ``incomplete`` after one read-back. A subsequent + invocation reads the remote set afresh and sends only truly missing edges. + Divergent edges and changed invitation/group/paper context always fail closed. + This guarantee depends on the injected transport's atomic create-only + operation; ordinary last-writer-wins bulk upsert is not a valid transport. + """ + + _verify_plan(plan) + if ( + isinstance(batch_size, bool) + or not isinstance(batch_size, int) + or not 1 <= batch_size <= _MAX_BATCH_SIZE + ): + raise DataValidationError(f"batch_size must be an integer between 1 and {_MAX_BATCH_SIZE}") + if not isinstance(publish, bool): + raise DataValidationError("publish must be a boolean") + if not publish: + return PublishReceipt(plan.digest, "dry_run", len(plan.edges), 0, 0) + if transport is None or confirm_sha256 != plan.digest: + raise DataValidationError( + "writing requires an injected transport and exact plan SHA-256 confirmation" + ) + _verify_context(plan.scope, transport.inspect_context(plan.scope)) + current = _remote_edges(plan, transport) + batches = 0 + while len(current) < len(plan.edges): + previous = current + _verify_context(plan.scope, transport.inspect_context(plan.scope)) + current = _remote_edges(plan, transport) + if not previous.issubset(current): + raise DataValidationError("remote edges disappeared during reconciliation") + pending = tuple(edge for edge in plan.edges if edge not in current) + if not pending: + return PublishReceipt(plan.digest, "complete", len(plan.edges), len(current), batches) + batch = pending[:batch_size] + batches += 1 + failed = False + try: + transport.post_edges(batch) + except Exception: # An ambiguous partial write must be reconciled, not blindly retried. + failed = True + try: + _verify_context(plan.scope, transport.inspect_context(plan.scope)) + current = _remote_edges(plan, transport) + except Exception: + return PublishReceipt( + plan.digest, + "incomplete", + len(plan.edges), + len(current), + batches, + "readback_unavailable", + ) + if not previous.issubset(current): + return PublishReceipt( + plan.digest, + "incomplete", + len(plan.edges), + len(current), + batches, + "unconfirmed_batch", + ) + if failed or not set(batch).issubset(current): + return PublishReceipt( + plan.digest, + "incomplete", + len(plan.edges), + len(current), + batches, + "unconfirmed_batch", + ) + return PublishReceipt(plan.digest, "complete", len(plan.edges), len(current), batches) + + +def write_publish_artifact(plan: PublishPlan, receipt: PublishReceipt, path: str | Path) -> None: + """Atomically link one evidence file to a destination that must not exist.""" + + _verify_plan(plan) + if receipt.plan_sha256 != plan.digest: + raise DataValidationError("receipt does not belong to this plan") + if receipt.expected_edges != len(plan.edges): + raise DataValidationError("receipt expected edge count does not match the plan") + destination = Path(path) + destination.parent.mkdir(parents=True, exist_ok=True) + body: dict[str, object] = { + "schema": "peermatchlab-openreview-publish-artifact-v1", + "plan": plan.as_dict(), + "receipt": receipt.as_dict(), + } + handle, temporary_name = tempfile.mkstemp( + prefix=f".{destination.name}-", suffix=".tmp", dir=destination.parent + ) + staging = Path(temporary_name) + try: + with os.fdopen(handle, "wb") as output: + output.write(_canonical_bytes(body)) + output.flush() + os.fsync(output.fileno()) + os.link(staging, destination) + finally: + if staging.exists(): + staging.unlink() diff --git a/tests/test_openreview_publish.py b/tests/test_openreview_publish.py new file mode 100644 index 0000000..a74afcc --- /dev/null +++ b/tests/test_openreview_publish.py @@ -0,0 +1,547 @@ +"""No test in this module contacts or writes to OpenReview.""" + +from __future__ import annotations + +import hashlib +import json +import os +from dataclasses import replace +from itertools import count +from pathlib import Path + +import pytest + +import peermatchlab.openreview_publish as publisher +from peermatchlab.models import ( + Assignment, + Conflict, + DataValidationError, + Document, + Expert, + MatchPlan, +) +from peermatchlab.openreview_publish import ( + AssignmentEdge, + PublishPlan, + PublishScope, + RemoteContext, + prepare_publish_plan, + publish_assignment_plan, + write_publish_artifact, +) + + +def fixture() -> tuple[MatchPlan, PublishScope, tuple[Document, ...], tuple[Expert, ...]]: + documents = ( + Document("paper-A", "A", required_experts=1), + Document("paper-B", "B", required_experts=1), + ) + experts = (Expert("~Alice1", "Alice", capacity=1), Expert("~Bob1", "Bob", capacity=1)) + plan = MatchPlan( + assignments=( + Assignment("paper-A", "~Alice1", 0.8, 1), + Assignment("paper-B", "~Bob1", 0.6, 1), + ), + unmet={"paper-A": 0, "paper-B": 0}, + strategy="optimal", + total_score=1.4, + ) + scope = PublishScope( + "Venue/-/Assignment", + "Venue/-/Aggregate_Score", + "Venue/Reviewers", + "peer-matchlab-v1", + ("paper-B", "paper-A"), + ("~Bob1", "~Alice1"), + ) + return plan, scope, documents, experts + + +def prepared() -> PublishPlan: + plan, scope, documents, experts = fixture() + return prepare_publish_plan(plan, scope, documents=documents, experts=experts, default_demand=1) + + +class MockTransport: + def __init__(self, plan: PublishPlan) -> None: + self.scope = plan.scope + self.remote: dict[tuple[str, str, str, str], AssignmentEdge] = {} + self.writes: list[tuple[AssignmentEdge, ...]] = [] + self.inspections = 0 + self.reads = 0 + self.fail_after: int | None = None + self.stale = False + + def inspect_context(self, scope: PublishScope) -> RemoteContext: + self.inspections += 1 + return RemoteContext( + scope.assignment_invitation, + scope.score_invitation, + scope.reviewer_group, + scope.paper_ids, + scope.reviewer_ids, + ) + + def read_edges(self, scope: PublishScope) -> tuple[AssignmentEdge, ...]: + self.reads += 1 + assert scope == self.scope + return tuple(self.remote.values()) + + def post_edges(self, edges: tuple[AssignmentEdge, ...]) -> None: + self.writes.append(edges) + if self.stale: + return + for index, edge in enumerate(edges): + if self.fail_after is not None and index == self.fail_after: + raise RuntimeError("simulated ambiguous transport failure") + if edge.key in self.remote: + raise RuntimeError("simulated conditional-insert conflict") + self.remote[edge.key] = edge + + +def test_dry_run_never_calls_even_an_injected_transport() -> None: + plan = prepared() + transport = MockTransport(plan) + result = publish_assignment_plan(plan, transport=transport) + assert result.status == "dry_run" + assert result.expected_edges == 4 + assert (transport.inspections, transport.reads, transport.writes) == (0, 0, []) + assert result.as_dict()["plan_sha256"] == plan.digest + assert publish_assignment_plan(plan).status == "dry_run" + + +def test_explicit_publish_reconciles_complete_remote_state_without_new_writes() -> None: + plan = prepared() + transport = MockTransport(plan) + with pytest.raises(DataValidationError, match="exact plan SHA-256"): + publish_assignment_plan(plan, transport=transport, publish=True) + assert not transport.writes and transport.inspections == 0 + result = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest, batch_size=2 + ) + assert (result.status, result.confirmed_edges, result.attempted_batches) == ("complete", 4, 2) + assert all(len(batch) <= 2 for batch in transport.writes) + assert set(transport.remote.values()) == set(plan.edges) + writes = len(transport.writes) + again = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest + ) + assert again.status == "complete" and again.attempted_batches == 0 + assert len(transport.writes) == writes + + +def rehashed(plan: PublishPlan) -> PublishPlan: + body = plan.as_dict() + body.pop("sha256") + return replace(plan, digest=hashlib.sha256(publisher._canonical_bytes(body)).hexdigest()) + + +def test_partial_transport_failure_returns_incomplete_and_retries_only_missing_edges() -> None: + plan = prepared() + transport = MockTransport(plan) + transport.fail_after = 1 + first = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest + ) + assert (first.status, first.confirmed_edges, first.attempted_batches, first.failure) == ( + "incomplete", + 1, + 1, + "unconfirmed_batch", + ) + first_key = next(iter(transport.remote)) + transport.fail_after = None + second = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest, batch_size=2 + ) + assert second.status == "complete" and second.confirmed_edges == 4 + assert all(edge.key != first_key for batch in transport.writes[1:] for edge in batch) + + +def test_concurrent_divergent_insert_between_read_and_post_is_never_overwritten() -> None: + plan = prepared() + transport = MockTransport(plan) + original_post = transport.post_edges + first = plan.edges[0] + divergent = replace(first, weight=0.11) + + def concurrent_insert(edges: tuple[AssignmentEdge, ...]) -> None: + transport.remote[first.key] = divergent + original_post(edges) + + transport.post_edges = concurrent_insert # type: ignore[method-assign] + result = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest + ) + assert result.status == "incomplete" and result.attempted_batches == 1 + assert transport.remote[first.key] == divergent + assert len(transport.remote) == 1 + with pytest.raises(DataValidationError, match="remote edge differs"): + publish_assignment_plan(plan, transport=transport, publish=True, confirm_sha256=plan.digest) + assert transport.remote[first.key] == divergent + + +def test_unconfirmed_write_stops_after_one_batch() -> None: + plan = prepared() + transport = MockTransport(plan) + transport.stale = True + result = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest, batch_size=1 + ) + assert result.status == "incomplete" and result.confirmed_edges == 0 + assert len(transport.writes) == 1 + + +def test_readback_failure_reports_uncertainty_and_retry_reconciles() -> None: + plan = prepared() + transport = MockTransport(plan) + original_read = transport.read_edges + + def intermittent_read(scope: PublishScope) -> tuple[AssignmentEdge, ...]: + if transport.writes: + raise RuntimeError("simulated unavailable read-back") + return original_read(scope) + + transport.read_edges = intermittent_read # type: ignore[method-assign] + first = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest, batch_size=2 + ) + assert (first.status, first.confirmed_edges, first.failure) == ( + "incomplete", + 0, + "readback_unavailable", + ) + assert len(transport.writes) == 1 + transport.read_edges = original_read # type: ignore[method-assign] + second = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest + ) + assert (second.status, second.confirmed_edges, second.attempted_batches) == ("complete", 4, 1) + assert set(transport.writes[1]) == set(plan.edges) - set(transport.writes[0]) + + +def test_context_drift_after_write_reports_incomplete() -> None: + plan = prepared() + transport = MockTransport(plan) + original_inspect = transport.inspect_context + + def drifting_context(scope: PublishScope) -> RemoteContext: + value = original_inspect(scope) + return replace(value, reviewer_group="changed") if transport.writes else value + + transport.inspect_context = drifting_context # type: ignore[method-assign] + result = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest, batch_size=1 + ) + assert (result.status, result.confirmed_edges, result.failure) == ( + "incomplete", + 0, + "readback_unavailable", + ) + assert len(transport.writes) == 1 + + +def test_remote_deletion_between_batches_never_causes_unbounded_reposting() -> None: + plan = prepared() + transport = MockTransport(plan) + original_post = transport.post_edges + + def replacing_post(edges: tuple[AssignmentEdge, ...]) -> None: + transport.remote.clear() + original_post(edges) + + transport.post_edges = replacing_post # type: ignore[method-assign] + receipt = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest, batch_size=1 + ) + assert receipt.status == "incomplete" and receipt.attempted_batches == 2 + assert len(transport.writes) == 2 + + +@pytest.mark.parametrize("case", ["wrong_weight", "unknown_edge", "duplicate"]) +def test_existing_divergent_or_duplicate_edges_fail_closed(case: str) -> None: + plan = prepared() + transport = MockTransport(plan) + edge = plan.edges[0] + if case == "wrong_weight": + transport.remote[edge.key] = replace(edge, weight=0.11) + elif case == "unknown_edge": + alien = replace(edge, tail="~Unknown") + transport.remote[alien.key] = alien + else: + transport.remote[edge.key] = edge + transport.read_edges = lambda scope: (edge, edge) # type: ignore[method-assign] + with pytest.raises(DataValidationError, match="remote edge"): + publish_assignment_plan(plan, transport=transport, publish=True, confirm_sha256=plan.digest) + assert not transport.writes + + +@pytest.mark.parametrize("case", ["wrong_type", "oversized", "malformed_row"]) +def test_untrusted_remote_edge_response_is_bounded_and_typed(case: str) -> None: + plan = prepared() + transport = MockTransport(plan) + if case == "wrong_type": + transport.read_edges = lambda scope: [] # type: ignore[method-assign] + elif case == "oversized": + transport.read_edges = lambda scope: (plan.edges[0],) * 2_001 # type: ignore[method-assign] + else: + transport.read_edges = lambda scope: ("not-an-edge",) # type: ignore[method-assign] + with pytest.raises(DataValidationError, match="remote edge response"): + publish_assignment_plan(plan, transport=transport, publish=True, confirm_sha256=plan.digest) + assert not transport.writes + + +def test_remote_context_must_be_typed_and_bounded() -> None: + plan = prepared() + transport = MockTransport(plan) + transport.inspect_context = lambda scope: {} # type: ignore[method-assign] + with pytest.raises(DataValidationError, match="remote context"): + publish_assignment_plan(plan, transport=transport, publish=True, confirm_sha256=plan.digest) + assert not transport.writes + + +def test_context_drift_fails_before_any_write() -> None: + plan = prepared() + transport = MockTransport(plan) + transport.inspect_context = lambda scope: RemoteContext( # type: ignore[method-assign] + scope.assignment_invitation, + "other-score-invitation", + scope.reviewer_group, + scope.paper_ids, + scope.reviewer_ids, + ) + with pytest.raises(DataValidationError, match="remote invitations"): + publish_assignment_plan(plan, transport=transport, publish=True, confirm_sha256=plan.digest) + assert not transport.writes + + +def test_prepare_rejects_unsafe_incomplete_or_wrong_scope() -> None: + plan, scope, documents, experts = fixture() + with pytest.raises(DataValidationError, match="paper_ids"): + prepare_publish_plan( + plan, + replace(scope, paper_ids=("paper-A",)), + documents=documents, + experts=experts, + default_demand=1, + ) + with pytest.raises(DataValidationError, match="only independently audited"): + prepare_publish_plan( + plan, + scope, + documents=documents, + experts=experts, + conflicts=(Conflict("paper-A", "~Alice1"),), + default_demand=1, + ) + incomplete = MatchPlan((plan.assignments[0],), {"paper-A": 0, "paper-B": 1}, "optimal", 0.8) + with pytest.raises(DataValidationError, match="only independently audited"): + prepare_publish_plan( + incomplete, scope, documents=documents, experts=experts, default_demand=1 + ) + + +def test_plan_digest_is_order_independent_but_binds_all_scoped_content() -> None: + plan, scope, documents, experts = fixture() + first = prepared() + reverse = MatchPlan( + tuple(reversed(plan.assignments)), plan.unmet, plan.strategy, plan.total_score + ) + assert ( + first.digest + == prepare_publish_plan( + reverse, scope, documents=documents, experts=experts, default_demand=1 + ).digest + ) + changed = replace(scope, label="a-different-run") + assert ( + first.digest + != prepare_publish_plan( + plan, changed, documents=documents, experts=experts, default_demand=1 + ).digest + ) + tampered = replace(first, edges=(replace(first.edges[0], weight=0.01), *first.edges[1:])) + with pytest.raises(DataValidationError, match="digest"): + publish_assignment_plan(tampered) + + +def test_rehashed_but_malformed_edge_sets_still_fail_closed() -> None: + plan = prepared() + malformed = ( + (replace(plan, edges=()), "invalid edge count"), + ( + replace(plan, edges=(replace(plan.edges[0], tail="~Alien"), *plan.edges[1:])), + "outside its exact scope", + ), + ( + replace(plan, edges=(replace(plan.edges[0], weight=0.11), *plan.edges[1:])), + "weights must agree", + ), + (replace(plan, edges=(plan.edges[0], plan.edges[0], *plan.edges[2:])), "exactly one edge"), + ) + for tampered, message in malformed: + with pytest.raises(DataValidationError, match=message): + publish_assignment_plan(rehashed(tampered)) + + +def test_duplicate_pair_is_rejected_before_edge_plan_is_emitted() -> None: + plan, scope, documents, experts = fixture() + duplicated = MatchPlan( + (plan.assignments[0], plan.assignments[0], plan.assignments[1]), + {"paper-A": 0, "paper-B": 0}, + "synthetic", + 2.2, + ) + with pytest.raises(DataValidationError, match="only independently audited"): + prepare_publish_plan( + duplicated, scope, documents=documents, experts=experts, default_demand=1 + ) + + +def test_peer_completes_plan_between_preflight_reads_without_empty_post() -> None: + plan = prepared() + transport = MockTransport(plan) + original_read = transport.read_edges + + def peer_completion(scope: PublishScope) -> tuple[AssignmentEdge, ...]: + if transport.reads == 1: + transport.remote = {edge.key: edge for edge in plan.edges} + return original_read(scope) + + transport.read_edges = peer_completion # type: ignore[method-assign] + receipt = publish_assignment_plan( + plan, transport=transport, publish=True, confirm_sha256=plan.digest + ) + assert receipt.status == "complete" and receipt.attempted_batches == 0 + assert not transport.writes + + +def test_bounds_and_ids_are_validated() -> None: + plan = prepared() + with pytest.raises(DataValidationError, match="batch_size"): + publish_assignment_plan(plan, batch_size=101) + with pytest.raises(DataValidationError, match="exact, printable"): + replace(plan.scope, reviewer_group="bad group") + with pytest.raises(DataValidationError, match="unique"): + replace(plan.scope, reviewer_ids=("~Alice1", "~Alice1")) + with pytest.raises(DataValidationError, match="finite score"): + replace(plan.edges[0], weight=float("nan")) + with pytest.raises(DataValidationError, match="finite score"): + replace(plan.edges[0], weight=True) + with pytest.raises(DataValidationError, match="publish must be"): + publish_assignment_plan(plan, publish=1) # type: ignore[arg-type] + with pytest.raises(DataValidationError, match="invitations must differ"): + replace(plan.scope, score_invitation=plan.scope.assignment_invitation) + + +def test_receipt_cannot_claim_unverified_or_invalid_state() -> None: + plan = prepared() + receipt = publish_assignment_plan(plan) + invalid = ( + ({"plan_sha256": "wrong"}, "lowercase SHA-256"), + ({"status": "bogus"}, "status is unknown"), + ({"attempted_batches": -1}, "non-negative integer"), + ({"expected_edges": 2_001}, "edge counts exceed"), + ({"status": "incomplete"}, "only incomplete"), + ({"failure": "unknown"}, "only incomplete"), + ({"confirmed_edges": 1}, "dry-run receipt"), + ) + for changes, message in invalid: + with pytest.raises(DataValidationError, match=message): + replace(receipt, **changes) + + +def test_unbounded_generators_are_stopped_at_the_declared_ceiling() -> None: + plan, scope, documents, experts = fixture() + seen = 0 + + def endless_ids(): + nonlocal seen + for index in count(): + seen += 1 + yield f"paper-{index}" + + with pytest.raises(DataValidationError, match="1 to 1000"): + replace(scope, paper_ids=endless_ids()) + assert seen == 1_001 + + seen = 0 + + def endless_documents(): + nonlocal seen + for _ in count(): + seen += 1 + yield documents[0] + + with pytest.raises(DataValidationError, match="documents exceeds maximum"): + prepare_publish_plan( + plan, scope, documents=endless_documents(), experts=experts, default_demand=1 + ) + assert seen == 1_001 + + seen = 0 + + def endless_experts(): + nonlocal seen + for _ in count(): + seen += 1 + yield experts[0] + + with pytest.raises(DataValidationError, match="experts exceeds maximum"): + prepare_publish_plan( + plan, scope, documents=documents, experts=endless_experts(), default_demand=1 + ) + assert seen == 10_001 + + seen = 0 + + def endless_conflicts(): + nonlocal seen + for _ in count(): + seen += 1 + yield Conflict("paper-A", "~Bob1") + + with pytest.raises(DataValidationError, match="conflicts exceeds maximum"): + prepare_publish_plan( + plan, + scope, + documents=documents, + experts=experts, + conflicts=endless_conflicts(), + default_demand=1, + ) + assert seen == 100_001 + + +def test_no_overwrite_local_artifact_contains_same_digest(tmp_path: Path) -> None: + plan = prepared() + receipt = publish_assignment_plan(plan) + output = tmp_path / "publish-evidence.json" + write_publish_artifact(plan, receipt, output) + artifact = json.loads(output.read_text(encoding="utf-8")) + assert artifact["plan"]["sha256"] == plan.digest + assert artifact["receipt"]["status"] == "dry_run" + with pytest.raises(FileExistsError): + write_publish_artifact(plan, receipt, output) + assert len(list(tmp_path.iterdir())) == 1 + with pytest.raises(DataValidationError, match="expected edge count"): + write_publish_artifact(plan, replace(receipt, expected_edges=2), tmp_path / "other") + with pytest.raises(DataValidationError, match="complete receipt"): + replace(receipt, status="complete") + + +def test_concurrent_destination_creation_is_never_replaced(tmp_path: Path, monkeypatch) -> None: + plan = prepared() + receipt = publish_assignment_plan(plan) + output = tmp_path / "race.json" + original_link = os.link + + def create_target_before_link(source: Path, destination: Path) -> None: + output.write_text("created concurrently", encoding="utf-8") + original_link(source, destination) + + monkeypatch.setattr(publisher.os, "link", create_target_before_link) + with pytest.raises(FileExistsError): + write_publish_artifact(plan, receipt, output) + assert output.read_text(encoding="utf-8") == "created concurrently" + assert list(tmp_path.iterdir()) == [output] diff --git a/uv.lock b/uv.lock index 9064cd4..52bf51f 100644 --- a/uv.lock +++ b/uv.lock @@ -478,7 +478,7 @@ wheels = [ [[package]] name = "peermatchlab" -version = "0.12.0" +version = "0.13.0" source = { editable = "." } [package.optional-dependencies]