added a review page functionality

README.md

@@ -5,21 +5,9 @@ mapping workflows.
 
 ## Current status
 
-The package provides source-independent Pydantic contracts, SQLAlchemy models,
-and a resumable `MappingStore` for runs, inputs, candidates, separate evidence
-records, versioned decisions, decision history, provenance, and lifecycle
-state. It can use an explicit SQLAlchemy URL for tests or resolve a named
-database resource through `oa-configurator` for consuming packages.
+The package provides source-independent Pydantic contracts, SQLAlchemy models, and a resumable `MappingStore` for runs, inputs, candidates, separate evidence records, versioned decisions, decision history, provenance, and lifecycle state. It can use an explicit SQLAlchemy URL for tests or resolve a named database resource through `oa-configurator` for consuming packages.
 
-`MappingReadContext` exposes common read-only status, coverage, evidence-packet,
-and review-handoff operations. Packets are JSON-safe and carry the full run,
-input, candidate, evidence, and decision-history lineage, so Groundworkers or a
-standalone reviewer can consume the same handoff without importing SQLAlchemy
-models.
-
-Representative contract fixtures under `tests/fixtures/` cover one-to-one,
-one-to-many, ambiguous, redirected/incomplete, unmappable, and retryable
-failure outcomes.
+`MappingReadContext` exposes common read-only status, coverage, paginated review summaries, evidence-packet, and review-handoff operations. Packets are JSON-safe and carry the full run, input, candidate, evidence, and decision-history lineage, so Groundworkers or a standalone reviewer can consume the same handoff without importing SQLAlchemy models. Review pages are query-level paginated and accept source-defined decision-status filters; `pending` is synthesized for inputs without a decision.
 
 ## Development
 
@@ -29,6 +17,3 @@ uv run ty check src/
 uv run ruff check .
 uv run pytest -q
 ```
-
-The package version is derived from a `vX.Y.Z` or `X.Y.Z` Git tag, with a
-`0.1.0` fallback when Git metadata is unavailable.

src/groundstore/__init__.py

@@ -1,6 +1,11 @@
 """Shared mapping-task contracts and persistence for Groundworkers workflows."""
 
-from .context import MappingEvidencePacket, MappingReadContext, MappingReviewHandoff
+from .context import (
+    MappingEvidencePacket,
+    MappingReadContext,
+    MappingReviewHandoff,
+    MappingReviewPage,
+)
 from .contracts import (
     DecisionStatus,
     LifecycleStatus,
@@ -23,6 +28,7 @@
     "MappingInputSpec",
     "MappingReadContext",
     "MappingReviewHandoff",
+    "MappingReviewPage",
     "MappingRunSpec",
     "MappingStore",
     "create_groundstore_engine",

src/groundstore/context.py

@@ -53,6 +53,21 @@ class MappingReviewHandoff(BaseModel):
     metadata: dict[str, Any] = Field(default_factory=dict)
 
 
+class MappingReviewPage(BaseModel):
+    """Stable, paginated summary for selecting mapping inputs to review."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: str = "groundstore.mapping-review-page.v1"
+    run: dict[str, Any] | None = None
+    page: int = Field(ge=1)
+    page_size: int = Field(ge=1)
+    page_count: int = Field(ge=0)
+    total_items: int = Field(ge=0)
+    decision_status: str | None = None
+    items: list[dict[str, Any]] = Field(default_factory=list)
+
+
 @dataclass(frozen=True, slots=True)
 class MappingReadContext:
     """Read-only Groundstore façade for host tools and review consumers."""
@@ -128,6 +143,73 @@ def evidence_packet(self, input_id: str) -> MappingEvidencePacket:
             decision=decision_payload,
         )
 
+    def review_page(
+        self,
+        source_namespace: str,
+        *,
+        run_id: str | None = None,
+        target_system: str | None = None,
+        page: int = 1,
+        page_size: int = 20,
+        decision_status: str | None = None,
+    ) -> MappingReviewPage:
+        """Return a database-paginated review summary for one mapping run."""
+        if page < 1:
+            raise ValueError("page must be positive")
+        if page_size < 1:
+            raise ValueError("page_size must be positive")
+
+        run = self.store.get_run(run_id) if run_id is not None else None
+        if run_id is None:
+            run = self.store.latest_successful_run(
+                source_namespace, target_system=target_system
+            )
+        if run is None:
+            return MappingReviewPage(
+                run=None,
+                page=page,
+                page_size=page_size,
+                page_count=0,
+                total_items=0,
+                decision_status=decision_status,
+            )
+        if run.source_namespace != source_namespace:
+            raise ValueError(f"mapping run {run.id} does not belong to {source_namespace}")
+        if target_system is not None and run.target_system != target_system:
+            raise ValueError(f"mapping run {run.id} does not target {target_system}")
+
+        records, total = self.store.get_review_inputs(
+            run.id,
+            offset=(page - 1) * page_size,
+            limit=page_size,
+            decision_status=decision_status,
+        )
+        page_count = (total + page_size - 1) // page_size
+        return MappingReviewPage(
+            run=_run_payload(run),
+            page=page,
+            page_size=page_size,
+            page_count=page_count,
+            total_items=total,
+            decision_status=decision_status,
+            items=[
+                {
+                    "input_id": input_record.id,
+                    "source_namespace": input_record.source_namespace,
+                    "source_kind": input_record.source_kind,
+                    "source_key": input_record.source_key,
+                    "lifecycle_status": input_record.lifecycle_status,
+                    "decision_status": status,
+                    "candidate_count": len(candidates),
+                    "candidates": [
+                        _review_candidate_payload(candidate) for candidate in candidates
+                    ],
+                    "normalized_projection": input_record.normalized_projection,
+                }
+                for input_record, status, candidates in records
+            ],
+        )
+
     def review_handoff(
         self,
         input_id: str,
@@ -207,6 +289,26 @@ def _candidate_payload(candidate: MappingCandidate) -> dict[str, Any]:
     }
 
 
+def _review_candidate_payload(candidate: MappingCandidate) -> dict[str, Any]:
+    payload = _candidate_payload(candidate)
+    return {
+        key: payload[key]
+        for key in (
+            "id",
+            "target_namespace",
+            "target_vocabulary_id",
+            "target_concept_id",
+            "target_code",
+            "target_grain",
+            "target_role",
+            "method",
+            "rank",
+            "score",
+            "confidence",
+        )
+    }
+
+
 def _evidence_payload(evidence: MappingEvidence) -> dict[str, Any]:
     return {
         "id": evidence.id,

src/groundstore/store.py

@@ -10,8 +10,8 @@
 from uuid import uuid4
 
 from oa_configurator import ResolvedDatabase, Resolver
-from sqlalchemy import Engine, select
-from sqlalchemy.orm import sessionmaker
+from sqlalchemy import Engine, and_, func, select
+from sqlalchemy.orm import aliased, sessionmaker
 
 from .contracts import (
     MappingCandidateSpec,
@@ -362,6 +362,105 @@ def get_inputs(self, run_id: str) -> list[MappingInput]:
         with self._session_factory() as session:
             return list(session.scalars(select(MappingInput).where(MappingInput.run_id == run_id)))
 
+    def get_review_inputs(
+        self,
+        run_id: str,
+        *,
+        offset: int = 0,
+        limit: int = 20,
+        decision_status: str | None = None,
+    ) -> tuple[list[tuple[MappingInput, str, list[MappingCandidate]]], int]:
+        """Return one stable, database-paginated review page.
+
+        The synthetic ``pending`` status represents an input without a
+        decision.  Other statuses are source-independent strings so adapters
+        may add meaningful outcomes without changing Groundstore's contract.
+        """
+        if offset < 0:
+            raise ValueError("offset cannot be negative")
+        if limit < 1:
+            raise ValueError("limit must be positive")
+
+        latest_versions = (
+            select(
+                MappingDecision.input_id,
+                func.max(MappingDecision.decision_version).label("decision_version"),
+            )
+            .group_by(MappingDecision.input_id)
+            .subquery()
+        )
+        latest_decision = aliased(MappingDecision)
+        decision_status_expression = func.coalesce(latest_decision.decision_status, "pending")
+        with self._session_factory() as session:
+            count_query = (
+                select(func.count(MappingInput.id))
+                .select_from(MappingInput)
+                .outerjoin(
+                    latest_versions,
+                    latest_versions.c.input_id == MappingInput.id,
+                )
+                .outerjoin(
+                    latest_decision,
+                    and_(
+                        latest_decision.input_id == MappingInput.id,
+                        latest_decision.decision_version
+                        == latest_versions.c.decision_version,
+                    ),
+                )
+                .where(MappingInput.run_id == run_id)
+            )
+            page_query = (
+                select(MappingInput, decision_status_expression)
+                .select_from(MappingInput)
+                .outerjoin(
+                    latest_versions,
+                    latest_versions.c.input_id == MappingInput.id,
+                )
+                .outerjoin(
+                    latest_decision,
+                    and_(
+                        latest_decision.input_id == MappingInput.id,
+                        latest_decision.decision_version
+                        == latest_versions.c.decision_version,
+                    ),
+                )
+                .where(MappingInput.run_id == run_id)
+                .order_by(
+                    MappingInput.source_kind,
+                    MappingInput.source_key,
+                    MappingInput.id,
+                )
+                .offset(offset)
+                .limit(limit)
+            )
+            if decision_status is not None:
+                count_query = count_query.where(decision_status_expression == decision_status)
+                page_query = page_query.where(decision_status_expression == decision_status)
+
+            total = int(session.scalar(count_query) or 0)
+            rows = list(session.execute(page_query))
+            input_ids = [input_record.id for input_record, _ in rows]
+            candidates_by_input: dict[str, list[MappingCandidate]] = {
+                input_id: [] for input_id in input_ids
+            }
+            if input_ids:
+                candidates = session.scalars(
+                    select(MappingCandidate)
+                    .where(MappingCandidate.input_id.in_(input_ids))
+                    .order_by(MappingCandidate.input_id, MappingCandidate.rank)
+                )
+                for candidate in candidates:
+                    candidates_by_input[candidate.input_id].append(candidate)
+
+            return [
+                (
+                    input_record,
+                    str(status),
+                    candidates_by_input[input_record.id],
+                )
+                for input_record, status in rows
+            ], total
+
     def latest_successful_run(
         self, source_namespace: str, *, target_system: str | None = None
     ) -> MappingRun | None:

tests/test_context.py

@@ -79,3 +79,75 @@ def test_read_context_builds_json_safe_packet_and_review_handoff(store):
     assert handoff.source_namespace == "pbs"
     assert handoff.metadata == {"queue": "pbs"}
     assert handoff.model_dump(mode="json")["packet"] == payload
+
+
+def test_read_context_paginates_review_inputs_and_filters_latest_status(store):
+    run = store.get_or_create_run(
+        MappingRunSpec(
+            source_namespace="pbs",
+            source_fingerprint="snapshot-page",
+            target_system="omop",
+            algorithm_version="mapper-1",
+            policy_version="policy-1",
+        )
+    )
+    for key in ("B", "A", "C"):
+        input_record = store.upsert_input(
+            run.id,
+            MappingInputSpec(
+                source_namespace="pbs",
+                source_kind="drug",
+                source_key=key,
+                source_fingerprint=f"input-{key}",
+                normalized_projection={"name": key},
+            ),
+        )
+        candidate = store.upsert_candidate(
+            input_record.id,
+            MappingCandidateSpec(
+                target_namespace="omop",
+                target_vocabulary_id="RxNorm",
+                target_concept_id=key,
+                target_grain="ingredient",
+                method="exact",
+                rank=1,
+            ),
+        )
+        store.record_decision(
+            input_record.id,
+            MappingDecisionSpec(
+                decision_status=(
+                    DecisionStatus.AMBIGUOUS if key != "B" else DecisionStatus.MAPPED
+                ),
+                selected_candidate_ids=[candidate.id],
+                outcome_code="pbs.test",
+            ),
+        )
+    store.upsert_input(
+        run.id,
+        MappingInputSpec(
+            source_namespace="pbs",
+            source_kind="drug",
+            source_key="D",
+            source_fingerprint="input-D",
+            normalized_projection={"name": "D"},
+        ),
+    )
+    store.update_run(run.id, lifecycle_status="complete")
+
+    context = MappingReadContext(store)
+    page = context.review_page(
+        "pbs", run_id=run.id, page=1, page_size=1, decision_status="ambiguous"
+    )
+    payload = page.model_dump(mode="json")
+
+    assert payload["schema_version"] == "groundstore.mapping-review-page.v1"
+    assert payload["total_items"] == 2
+    assert payload["page_count"] == 2
+    assert payload["items"][0]["source_key"] == "A"
+    assert payload["items"][0]["candidate_count"] == 1
+    assert payload["items"][0]["decision_status"] == "ambiguous"
+
+    pending = context.review_page("pbs", run_id=run.id, decision_status="pending")
+    assert pending.total_items == 1
+    assert pending.items[0]["source_key"] == "D"