feat(review): add multi-owner contracts and receipts

Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d

api/main.py

@@ -23,6 +23,7 @@ from api.routers import reconciliation
 from api.routers import execution
 from api.routers import fabric
 from api.routers import legacy_meter
+from api.routers import review_contracts
 
 
 class ETagMiddleware(BaseHTTPMiddleware):
@@ -134,6 +135,7 @@ app.include_router(reconciliation.router)
 app.include_router(execution.router)
 app.include_router(fabric.router)
 app.include_router(legacy_meter.router)
+app.include_router(review_contracts.router)
 app.include_router(state.router)
 app.include_router(ops_runs.router)
 app.include_router(policy.router)

api/models/__init__.py

@@ -42,6 +42,7 @@ from api.models.fabric_graph import FabricGraphImport, FabricGraphNode, FabricGr
 from api.models.legacy_meter import LegacyInterface, LegacyInterfaceUsageBucket
 from api.models.write_idempotency_key import WriteIdempotencyKey
 from api.models.work_record_identifier_alias import WorkRecordIdentifierAlias
+from api.models.review_contract import ReviewContract, ReviewReceipt
 from api.models.suggestion import (
     Suggestion,
     SuggestionNote,
@@ -83,5 +84,6 @@ __all__ = [
     "LegacyInterface", "LegacyInterfaceUsageBucket",
     "WriteIdempotencyKey",
     "WorkRecordIdentifierAlias",
+    "ReviewContract", "ReviewReceipt",
     "Suggestion", "SuggestionNote", "SuggestionRelevanceBump", "SuggestionStage",
 ]

api/models/review_contract.py

@@ -0,0 +1,128 @@
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    ForeignKey,
+    Index,
+    String,
+    Text,
+    UniqueConstraint,
+)
+from sqlalchemy.dialects.postgresql import JSONB, UUID
+from sqlalchemy.orm import Mapped, mapped_column, relationship
+
+from api.models.base import Base, TimestampMixin, new_uuid7
+
+
+class ReviewContract(Base, TimestampMixin):
+    """Replaceable projection of one authoritative review-contract revision."""
+
+    __tablename__ = "review_contracts"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid7
+    )
+    contract_key: Mapped[str] = mapped_column(String(160), nullable=False, index=True)
+    schema_version: Mapped[str] = mapped_column(String(40), nullable=False)
+    contract_digest: Mapped[str] = mapped_column(
+        String(64), nullable=False, unique=True
+    )
+    source_repo: Mapped[str] = mapped_column(String(100), nullable=False)
+    source_path: Mapped[str] = mapped_column(Text, nullable=False)
+    source_revision: Mapped[str] = mapped_column(String(64), nullable=False)
+    document: Mapped[dict] = mapped_column(JSONB, nullable=False)
+    active: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=True, index=True
+    )
+    required_for_decision: Mapped[bool] = mapped_column(
+        Boolean, nullable=False, default=False
+    )
+    decision_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("decisions.id", ondelete="RESTRICT"),
+        nullable=True,
+        index=True,
+    )
+    workplan_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("workplans.id", ondelete="RESTRICT", onupdate="CASCADE"),
+        nullable=True,
+        index=True,
+    )
+    task_id: Mapped[uuid.UUID | None] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("tasks.id", ondelete="RESTRICT", onupdate="CASCADE"),
+        nullable=True,
+        index=True,
+    )
+    projected_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False
+    )
+
+    receipts: Mapped[list["ReviewReceipt"]] = relationship(
+        "ReviewReceipt", back_populates="contract", lazy="selectin"
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "contract_key",
+            "source_repo",
+            "source_path",
+            "source_revision",
+            name="uq_review_contract_source_revision",
+        ),
+        Index("ix_review_contract_key_active", "contract_key", "active"),
+    )
+
+
+class ReviewReceipt(Base, TimestampMixin):
+    """Immutable evidence submitted against an exact contract digest."""
+
+    __tablename__ = "review_receipts"
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True), primary_key=True, default=new_uuid7
+    )
+    contract_id: Mapped[uuid.UUID] = mapped_column(
+        UUID(as_uuid=True),
+        ForeignKey("review_contracts.id", ondelete="RESTRICT"),
+        nullable=False,
+        index=True,
+    )
+    owner_id: Mapped[str] = mapped_column(String(160), nullable=False, index=True)
+    actor: Mapped[str] = mapped_column(String(160), nullable=False)
+    disposition: Mapped[str] = mapped_column(String(32), nullable=False)
+    contract_digest: Mapped[str] = mapped_column(String(64), nullable=False, index=True)
+    receipt_digest: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
+    artifact_hashes: Mapped[dict] = mapped_column(JSONB, nullable=False, default=dict)
+    assertion_ids: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
+    checks: Mapped[list] = mapped_column(JSONB, nullable=False, default=list)
+    note: Mapped[str | None] = mapped_column(Text, nullable=True)
+    source_repo: Mapped[str] = mapped_column(String(100), nullable=False)
+    source_path: Mapped[str] = mapped_column(Text, nullable=False)
+    source_revision: Mapped[str] = mapped_column(String(64), nullable=False)
+    submitted_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, index=True
+    )
+    document: Mapped[dict] = mapped_column(JSONB, nullable=False)
+
+    contract: Mapped[ReviewContract] = relationship(
+        "ReviewContract", back_populates="receipts"
+    )
+
+    __table_args__ = (
+        UniqueConstraint(
+            "source_repo",
+            "source_path",
+            "source_revision",
+            name="uq_review_receipt_source_revision",
+        ),
+        Index(
+            "ix_review_receipt_contract_owner_time",
+            "contract_id",
+            "owner_id",
+            "submitted_at",
+        ),
+    )

api/routers/decisions.py

@@ -6,7 +6,6 @@ from pathlib import Path
 
 from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
 
-logger = logging.getLogger(__name__)
 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
@@ -14,9 +13,12 @@ from api.database import get_session
 from api.events import EventEnvelope, publish_event
 from api.models.decision import Decision, DecisionStatus, DecisionType
 from api.models.progress_event import ProgressEvent
+from api.models.review_contract import ReviewContract
 from api.schemas.decision import DecisionCreate, DecisionRead, DecisionResolve, DecisionUpdate
 from api.services.legacy_compat import meter_legacy_body_from_model, meter_legacy_query_param
+from api.services.review_contracts import aggregate
 
+logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/decisions", tags=["decisions"])
 
 _FINANCIAL_LEGAL_KEYWORDS = (
@@ -154,6 +156,25 @@ async def resolve_decision_action(
     if decision.status == DecisionStatus.resolved:
         raise HTTPException(status_code=409, detail="Decision already resolved")
 
+    review_rows = await session.execute(
+        select(ReviewContract).where(
+            ReviewContract.decision_id == decision.id,
+            ReviewContract.active.is_(True),
+            ReviewContract.required_for_decision.is_(True),
+        )
+    )
+    for contract in review_rows.scalars():
+        review_state = await aggregate(session, contract)
+        if not review_state.satisfied:
+            raise HTTPException(
+                status_code=409,
+                detail={
+                    "message": "required multi-owner review is not satisfied",
+                    "contract_key": contract.contract_key,
+                    "contract_digest": contract.contract_digest,
+                },
+            )
+
     decision.status = DecisionStatus.resolved
     decision.decision_type = DecisionType.made
     decision.rationale = body.rationale

api/routers/review_contracts.py

@@ -0,0 +1,190 @@
+import uuid
+
+from fastapi import APIRouter, Depends, HTTPException, Query, status
+from sqlalchemy import select, update
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.database import get_session
+from api.models.review_contract import ReviewContract, ReviewReceipt
+from api.schemas.review_contract import (
+    ReviewAggregateRead,
+    ReviewContractProject,
+    ReviewContractRead,
+    ReviewReceiptRead,
+    ReviewReceiptSubmit,
+)
+from api.services.review_contracts import (
+    active_contract,
+    aggregate,
+    digest_document,
+    normalize_contract,
+    utc_now,
+    validate_receipt,
+)
+
+router = APIRouter(prefix="/review-contracts", tags=["review-contracts"])
+
+
+@router.post(
+    "/projections",
+    response_model=ReviewContractRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def project_contract(
+    body: ReviewContractProject,
+    session: AsyncSession = Depends(get_session),
+) -> ReviewContract:
+    document, contract_digest = normalize_contract(body.contract)
+    existing = (
+        await session.execute(
+            select(ReviewContract).where(
+                ReviewContract.contract_digest == contract_digest
+            )
+        )
+    ).scalar_one_or_none()
+    if existing:
+        if (
+            existing.source_repo != body.source.repo
+            or existing.source_path != body.source.path
+            or existing.source_revision != body.source.revision
+        ):
+            raise HTTPException(
+                409, "contract digest is already projected from a different source"
+            )
+        return existing
+    contract_key = document["contract_key"]
+    await session.execute(
+        update(ReviewContract)
+        .where(
+            ReviewContract.contract_key == contract_key, ReviewContract.active.is_(True)
+        )
+        .values(active=False)
+    )
+    contract = ReviewContract(
+        contract_key=contract_key,
+        schema_version="review-contract/v1",
+        contract_digest=contract_digest,
+        source_repo=body.source.repo,
+        source_path=body.source.path,
+        source_revision=body.source.revision,
+        document=document,
+        active=True,
+        required_for_decision=body.required_for_decision,
+        decision_id=body.decision_id,
+        workplan_id=body.workplan_id,
+        task_id=body.task_id,
+        projected_at=utc_now(),
+    )
+    session.add(contract)
+    try:
+        await session.commit()
+    except IntegrityError as exc:
+        await session.rollback()
+        raise HTTPException(
+            409, "source revision is already projected with different content"
+        ) from exc
+    await session.refresh(contract)
+    return contract
+
+
+@router.get("/", response_model=list[ReviewContractRead])
+async def list_contracts(
+    contract_key: str | None = Query(None),
+    active: bool | None = Query(None),
+    decision_id: uuid.UUID | None = Query(None),
+    session: AsyncSession = Depends(get_session),
+) -> list[ReviewContract]:
+    query = select(ReviewContract).order_by(ReviewContract.projected_at.desc())
+    if contract_key:
+        query = query.where(ReviewContract.contract_key == contract_key)
+    if active is not None:
+        query = query.where(ReviewContract.active == active)
+    if decision_id:
+        query = query.where(ReviewContract.decision_id == decision_id)
+    return list((await session.execute(query)).scalars())
+
+
+@router.get("/{contract_key}", response_model=ReviewContractRead)
+async def get_active_contract(
+    contract_key: str,
+    session: AsyncSession = Depends(get_session),
+) -> ReviewContract:
+    return await active_contract(session, contract_key)
+
+
+@router.get("/{contract_key}/aggregate", response_model=ReviewAggregateRead)
+async def get_aggregate(
+    contract_key: str,
+    session: AsyncSession = Depends(get_session),
+) -> ReviewAggregateRead:
+    return await aggregate(session, await active_contract(session, contract_key))
+
+
+@router.get("/{contract_key}/receipts", response_model=list[ReviewReceiptRead])
+async def list_receipts(
+    contract_key: str,
+    include_stale: bool = Query(True),
+    session: AsyncSession = Depends(get_session),
+) -> list[ReviewReceipt]:
+    contract = await active_contract(session, contract_key)
+    query = (
+        select(ReviewReceipt)
+        .join(ReviewContract)
+        .where(
+            ReviewContract.contract_key == contract_key
+            if include_stale
+            else ReviewReceipt.contract_id == contract.id
+        )
+        .order_by(ReviewReceipt.submitted_at)
+    )
+    return list((await session.execute(query)).scalars())
+
+
+@router.post(
+    "/{contract_key}/receipts",
+    response_model=ReviewReceiptRead,
+    status_code=status.HTTP_201_CREATED,
+)
+async def submit_receipt(
+    contract_key: str,
+    body: ReviewReceiptSubmit,
+    session: AsyncSession = Depends(get_session),
+) -> ReviewReceipt:
+    contract = await active_contract(session, contract_key)
+    document = validate_receipt(contract, body)
+    receipt_digest = digest_document(document)
+    existing = (
+        await session.execute(
+            select(ReviewReceipt).where(ReviewReceipt.receipt_digest == receipt_digest)
+        )
+    ).scalar_one_or_none()
+    if existing:
+        return existing
+    receipt = ReviewReceipt(
+        contract_id=contract.id,
+        owner_id=body.owner_id,
+        actor=body.actor,
+        disposition=body.disposition,
+        contract_digest=body.contract_digest,
+        receipt_digest=receipt_digest,
+        artifact_hashes=body.artifact_hashes,
+        assertion_ids=body.assertion_ids,
+        checks=body.checks,
+        note=body.note,
+        source_repo=body.source.repo,
+        source_path=body.source.path,
+        source_revision=body.source.revision,
+        submitted_at=utc_now(),
+        document=document,
+    )
+    session.add(receipt)
+    try:
+        await session.commit()
+    except IntegrityError as exc:
+        await session.rollback()
+        raise HTTPException(
+            409, "receipt source revision already contains different evidence"
+        ) from exc
+    await session.refresh(receipt)
+    return receipt

api/schemas/review_contract.py

@@ -0,0 +1,96 @@
+import uuid
+from datetime import datetime
+from typing import Any, Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+
+class SourceRef(BaseModel):
+    repo: str = Field(min_length=1, max_length=100)
+    path: str = Field(min_length=1, max_length=1000)
+    revision: str = Field(pattern=r"^[0-9a-f]{40,64}$")
+
+
+class ReviewContractProject(BaseModel):
+    contract: dict[str, Any]
+    source: SourceRef
+    decision_id: uuid.UUID | None = None
+    workplan_id: uuid.UUID | None = None
+    task_id: uuid.UUID | None = None
+    required_for_decision: bool = False
+
+
+class ReviewContractRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    contract_key: str
+    schema_version: str
+    contract_digest: str
+    source_repo: str
+    source_path: str
+    source_revision: str
+    document: dict[str, Any]
+    active: bool
+    required_for_decision: bool
+    decision_id: uuid.UUID | None
+    workplan_id: uuid.UUID | None
+    task_id: uuid.UUID | None
+    projected_at: datetime
+    created_at: datetime
+    updated_at: datetime
+
+
+class ReviewReceiptSubmit(BaseModel):
+    owner_id: str = Field(min_length=1, max_length=160)
+    actor: str = Field(min_length=2, max_length=160)
+    disposition: Literal["approve", "request_changes"]
+    contract_digest: str = Field(pattern=r"^[0-9a-f]{64}$")
+    artifact_hashes: dict[str, str] = Field(default_factory=dict)
+    assertion_ids: list[str] = Field(default_factory=list)
+    checks: list[dict[str, Any]] = Field(default_factory=list)
+    note: str | None = Field(default=None, max_length=2000)
+    source: SourceRef
+
+
+class ReviewReceiptRead(BaseModel):
+    model_config = ConfigDict(from_attributes=True)
+    id: uuid.UUID
+    contract_id: uuid.UUID
+    owner_id: str
+    actor: str
+    disposition: str
+    contract_digest: str
+    receipt_digest: str
+    artifact_hashes: dict[str, str]
+    assertion_ids: list[str]
+    checks: list[dict[str, Any]]
+    note: str | None
+    source_repo: str
+    source_path: str
+    source_revision: str
+    submitted_at: datetime
+    created_at: datetime
+
+
+class OwnerReviewState(BaseModel):
+    owner_id: str
+    status: Literal["missing", "approved", "request_changes", "stale"]
+    receipt_id: uuid.UUID | None = None
+    receipt_digest: str | None = None
+    submitted_at: datetime | None = None
+
+
+class GateReviewState(BaseModel):
+    gate_id: str
+    policy: Literal["all_required"]
+    owners: list[str]
+    satisfied: bool
+
+
+class ReviewAggregateRead(BaseModel):
+    contract_key: str
+    contract_digest: str
+    satisfied: bool
+    authorizes_execution: Literal[False] = False
+    owners: list[OwnerReviewState]
+    gates: list[GateReviewState]

api/services/review_contracts.py

@@ -0,0 +1,336 @@
+from __future__ import annotations
+
+import hashlib
+import json
+from datetime import datetime, timezone
+from typing import Any
+
+from fastapi import HTTPException
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from api.models.review_contract import ReviewContract, ReviewReceipt
+from api.schemas.review_contract import (
+    GateReviewState,
+    OwnerReviewState,
+    ReviewAggregateRead,
+)
+
+
+def canonical_json(value: Any) -> bytes:
+    return json.dumps(
+        value, sort_keys=True, separators=(",", ":"), ensure_ascii=True
+    ).encode()
+
+
+def digest_document(value: Any) -> str:
+    return hashlib.sha256(canonical_json(value)).hexdigest()
+
+
+def _assertion_id(text: str) -> str:
+    return "assertion:" + hashlib.sha256(text.encode()).hexdigest()[:20]
+
+
+def normalize_contract(raw: dict[str, Any]) -> tuple[dict[str, Any], str]:
+    """Validate v1 or adapt the WP-0024 prototype without executing its checks."""
+    digest = digest_document(raw)
+    if raw.get("interface") == "railiance.owner-review" and raw.get("version") == 1:
+        owners_raw = raw.get("owners")
+        task_owners = raw.get("task_owners")
+        hashes = raw.get("artifact_sha256")
+        if (
+            not isinstance(owners_raw, dict)
+            or not owners_raw
+            or not isinstance(task_owners, dict)
+            or not task_owners
+        ):
+            raise HTTPException(
+                422, "legacy contract requires non-empty owners and task_owners"
+            )
+        if not isinstance(hashes, dict):
+            raise HTTPException(422, "legacy contract requires artifact_sha256")
+        owners = []
+        for owner_id, scope in owners_raw.items():
+            artifacts = scope.get("artifacts") or []
+            assertions = scope.get("assertions") or []
+            checks = scope.get("checks") or []
+            if (
+                not artifacts
+                or not assertions
+                or not checks
+                or any(path not in hashes for path in artifacts)
+            ):
+                raise HTTPException(422, f"{owner_id}: incomplete legacy owner scope")
+            owners.append(
+                {
+                    "id": owner_id,
+                    "artifact_ids": artifacts,
+                    "assertions": [
+                        {"id": _assertion_id(text), "statement": text}
+                        for text in assertions
+                    ],
+                    "check_ids": checks,
+                }
+            )
+        owner_ids = {item["id"] for item in owners}
+        gates = []
+        for gate_id, required in task_owners.items():
+            if (
+                not required
+                or len(required) != len(set(required))
+                or not set(required) <= owner_ids
+            ):
+                raise HTTPException(422, f"{gate_id}: invalid legacy owner set")
+            gates.append({"id": gate_id, "policy": "all_required", "owners": required})
+        normalized = {
+            "schema_version": "review-contract/v1",
+            "contract_key": raw.get("workplan_id"),
+            "subject": {"kind": "workplan", "id": raw.get("workplan_id")},
+            "owners": owners,
+            "gates": gates,
+            "artifacts": {
+                path: {"algorithm": "sha256", "digest": value}
+                for path, value in hashes.items()
+            },
+            "allowed_dispositions": [
+                item.replace("-", "_") for item in (raw.get("decisions") or [])
+            ],
+            "legacy": {
+                "interface": raw["interface"],
+                "version": 1,
+                "source_digest": digest,
+            },
+        }
+        return validate_v1(normalized), digest
+    if raw.get("schema_version") != "review-contract/v1":
+        raise HTTPException(422, "schema_version must be review-contract/v1")
+    return validate_v1(raw), digest
+
+
+def validate_v1(doc: dict[str, Any]) -> dict[str, Any]:
+    contract_key, subject = doc.get("contract_key"), doc.get("subject")
+    owners, gates, artifacts = doc.get("owners"), doc.get("gates"), doc.get("artifacts")
+    dispositions = doc.get("allowed_dispositions")
+    if not isinstance(contract_key, str) or not contract_key:
+        raise HTTPException(422, "contract_key is required")
+    if (
+        not isinstance(subject, dict)
+        or not subject.get("kind")
+        or not subject.get("id")
+    ):
+        raise HTTPException(422, "typed subject kind and id are required")
+    if (
+        not isinstance(owners, list)
+        or not owners
+        or not isinstance(gates, list)
+        or not gates
+    ):
+        raise HTTPException(422, "non-empty owners and gates are required")
+    if not isinstance(artifacts, dict) or not artifacts:
+        raise HTTPException(422, "revision-pinned artifacts are required")
+    if (
+        set(dispositions or []) != {"approve", "request_changes"}
+        or len(dispositions) != 2
+    ):
+        raise HTTPException(
+            422, "allowed_dispositions must contain approve and request_changes only"
+        )
+    owner_ids: set[str] = set()
+    for owner in owners:
+        if not isinstance(owner, dict):
+            raise HTTPException(422, "owner entries must be objects")
+        owner_id = owner.get("id")
+        artifact_ids = owner.get("artifact_ids") or []
+        assertions = owner.get("assertions") or []
+        checks = owner.get("check_ids") or []
+        assertion_ids = [
+            item.get("id") for item in assertions if isinstance(item, dict)
+        ]
+        if (
+            not owner_id
+            or owner_id in owner_ids
+            or not artifact_ids
+            or not checks
+            or not assertion_ids
+        ):
+            raise HTTPException(
+                422,
+                "each owner needs a unique id and non-empty artifact/assertion/check scope",
+            )
+        if len(assertion_ids) != len(set(assertion_ids)) or any(
+            path not in artifacts for path in artifact_ids
+        ):
+            raise HTTPException(422, f"{owner_id}: invalid assertion or artifact scope")
+        owner_ids.add(owner_id)
+    for artifact_id, artifact in artifacts.items():
+        if (
+            not isinstance(artifact, dict)
+            or artifact.get("algorithm") != "sha256"
+            or not _is_sha256(artifact.get("digest"))
+        ):
+            raise HTTPException(
+                422, f"{artifact_id}: only sha256 artifacts are supported"
+            )
+    gate_ids: set[str] = set()
+    for gate in gates:
+        if not isinstance(gate, dict):
+            raise HTTPException(422, "gate entries must be objects")
+        required = gate.get("owners") or []
+        if (
+            gate.get("policy") != "all_required"
+            or not gate.get("id")
+            or gate["id"] in gate_ids
+        ):
+            raise HTTPException(
+                422, "v1 gates require unique ids and all_required policy"
+            )
+        if (
+            not required
+            or len(required) != len(set(required))
+            or not set(required) <= owner_ids
+        ):
+            raise HTTPException(422, f"{gate.get('id')}: gate owner set is invalid")
+        gate_ids.add(gate["id"])
+    return doc
+
+
+def _is_sha256(value: Any) -> bool:
+    return (
+        isinstance(value, str)
+        and len(value) == 64
+        and all(c in "0123456789abcdef" for c in value)
+    )
+
+
+def owner_scope(contract: ReviewContract, owner_id: str) -> dict[str, Any]:
+    for owner in contract.document["owners"]:
+        if owner["id"] == owner_id:
+            return owner
+    raise HTTPException(422, f"unknown owner: {owner_id}")
+
+
+def validate_receipt(contract: ReviewContract, body: Any) -> dict[str, Any]:
+    if body.contract_digest != contract.contract_digest:
+        raise HTTPException(
+            409, "contract digest mismatch; re-review the active contract"
+        )
+    scope = owner_scope(contract, body.owner_id)
+    if body.disposition == "request_changes":
+        if not body.note or not body.note.strip():
+            raise HTTPException(422, "request_changes requires a note")
+    else:
+        expected_artifacts = {
+            key: contract.document["artifacts"][key]["digest"]
+            for key in scope["artifact_ids"]
+        }
+        if body.artifact_hashes != expected_artifacts:
+            raise HTTPException(
+                422, "approval artifact hashes do not exactly match the owner scope"
+            )
+        expected_assertions = {item["id"] for item in scope["assertions"]}
+        if set(body.assertion_ids) != expected_assertions or len(
+            body.assertion_ids
+        ) != len(expected_assertions):
+            raise HTTPException(
+                422, "approval assertions do not exactly match the owner scope"
+            )
+        expected_checks = set(scope["check_ids"])
+        actual_checks = {
+            item.get("id") for item in body.checks if isinstance(item, dict)
+        }
+        if actual_checks != expected_checks or len(body.checks) != len(expected_checks):
+            raise HTTPException(
+                422, "approval checks do not exactly match the owner scope"
+            )
+        if any(
+            item.get("passed") is not True or item.get("read_only") is not True
+            for item in body.checks
+        ):
+            raise HTTPException(
+                422, "every approval check must be passed and explicitly read_only"
+            )
+    return {
+        "owner_id": body.owner_id,
+        "actor": body.actor,
+        "disposition": body.disposition,
+        "contract_digest": body.contract_digest,
+        "artifact_hashes": body.artifact_hashes,
+        "assertion_ids": body.assertion_ids,
+        "checks": body.checks,
+        "note": body.note,
+        "source": body.source.model_dump(),
+    }
+
+
+async def active_contract(session: AsyncSession, contract_key: str) -> ReviewContract:
+    result = await session.execute(
+        select(ReviewContract).where(
+            ReviewContract.contract_key == contract_key, ReviewContract.active.is_(True)
+        )
+    )
+    contract = result.scalar_one_or_none()
+    if contract is None:
+        raise HTTPException(404, f"active review contract not found: {contract_key}")
+    return contract
+
+
+async def aggregate(
+    session: AsyncSession, contract: ReviewContract
+) -> ReviewAggregateRead:
+    rows = await session.execute(
+        select(ReviewReceipt)
+        .where(ReviewReceipt.contract_id == contract.id)
+        .order_by(ReviewReceipt.submitted_at.desc(), ReviewReceipt.id.desc())
+    )
+    latest: dict[str, ReviewReceipt] = {}
+    for receipt in rows.scalars():
+        latest.setdefault(receipt.owner_id, receipt)
+    prior = await session.execute(
+        select(ReviewReceipt.owner_id)
+        .join(ReviewContract)
+        .where(
+            ReviewContract.contract_key == contract.contract_key,
+            ReviewContract.id != contract.id,
+        )
+        .distinct()
+    )
+    stale_owners = set(prior.scalars())
+    owner_states, statuses = [], {}
+    for owner in contract.document["owners"]:
+        owner_id, receipt = owner["id"], latest.get(owner["id"])
+        state = (
+            receipt.disposition.replace("approve", "approved")
+            if receipt
+            else ("stale" if owner_id in stale_owners else "missing")
+        )
+        statuses[owner_id] = state
+        owner_states.append(
+            OwnerReviewState(
+                owner_id=owner_id,
+                status=state,
+                receipt_id=receipt.id if receipt else None,
+                receipt_digest=receipt.receipt_digest if receipt else None,
+                submitted_at=receipt.submitted_at if receipt else None,
+            )
+        )
+    gates = [
+        GateReviewState(
+            gate_id=gate["id"],
+            policy="all_required",
+            owners=gate["owners"],
+            satisfied=all(statuses[owner] == "approved" for owner in gate["owners"]),
+        )
+        for gate in contract.document["gates"]
+    ]
+    return ReviewAggregateRead(
+        contract_key=contract.contract_key,
+        contract_digest=contract.contract_digest,
+        satisfied=all(gate.satisfied for gate in gates),
+        authorizes_execution=False,
+        owners=owner_states,
+        gates=gates,
+    )
+
+
+def utc_now() -> datetime:
+    return datetime.now(timezone.utc)