diff --git a/CHANGELOG.md b/CHANGELOG.md index e53670d..696a666 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,13 @@ ## Unreleased +- Freeze the `cfr-review-state/v1` human-review state machine. Independent + approve/reject decisions now compose in either order, approvals bind to their + immutable audit-trace prefix, exact retries are idempotent, and concurrent + mutations use fail-closed revision compare-and-set semantics. Admin-only JSON + review endpoints derive reviewer identity from the verified authorization + context; the public resolve surface and all retrieval behavior remain unchanged. + - Freeze the prospective `cfr-unit-fields/v1` contract: application quantities remain aligned to the factor denominator, while `resolved_quantity_kg` now means kilograms only and remains null for non-mass activity. Preview totals normalize `gCO2e`/`tCO2e` numerators to `kgCO2e`, and lock diff --git a/README.md b/README.md index d9aaffb..48feac9 100644 --- a/README.md +++ b/README.md @@ -223,6 +223,7 @@ keeps both fields outside the production response allowlist. - [Contributing](CONTRIBUTING.md) - [Bring Your Own Catalog](docs/BRING_YOUR_OWN_CATALOG.md) - [Catalog-to-lock integrity contract](docs/CFR_CATALOG_LOCK_INTEGRITY_V3.md) +- [Human-review workflow state machine](docs/CFR_REVIEW_WORKFLOW_STATE_MACHINE_V1.md) - [Public API safety contract](docs/CFR_API_SAFETY_CONTRACT.md) - [Release notes](docs/RELEASE_NOTES_V0.14.3.md) - [Technical implementation reference](docs/CFR_TECHNICAL_DOCUMENT.md) diff --git a/docs/CFR_REVIEW_WORKFLOW_STATE_MACHINE_V1.md b/docs/CFR_REVIEW_WORKFLOW_STATE_MACHINE_V1.md new file mode 100644 index 0000000..677e5eb --- /dev/null +++ b/docs/CFR_REVIEW_WORKFLOW_STATE_MACHINE_V1.md @@ -0,0 +1,86 @@ +# CFR Review Workflow State Machine v1 + +Status: **FROZEN** for PR-C (`codex/review-workflow-state-machine-v1`). + +This contract governs only human review decisions and immutable locking. It does +not change retrieval, ranking, semantic matching, candidate identity, catalog +content, factor values, qualification, or the public resolve API. + +## States + +A resolution run has one of three review states: + +- `OPEN`: no candidate has been approved and the resolution is not locked. +- `SELECTED`: exactly one candidate has an `APPROVED` decision. Other candidates + may independently have `REJECTED` decisions. +- `LOCKED`: the selected candidate and its approval have been committed to an + immutable `LockedResolution`. + +Each candidate has at most one terminal human decision in a resolution run: +`APPROVED` or `REJECTED`. A request can have at most one approved candidate and +at most one lock. + +## Legal transitions + +- `OPEN -> OPEN`: reject any undecided candidate. +- `OPEN -> SELECTED`: approve one undecided candidate. +- `SELECTED -> SELECTED`: reject any other undecided candidate. +- `SELECTED -> LOCKED`: lock the approved candidate. +- An exact replay of a committed approve or reject operation returns the + previously committed object without appending another trace event. A lock + retry for the already locked candidate returns the canonical immutable lock; + its original reviewer identity is never replaced by the retrying caller. + +The order of independent decisions does not change lockability. In particular, +`approve A -> reject B -> lock A` and `reject B -> approve A -> lock A` are both +valid. Their append-only audit traces may differ in event order, while the same +candidate and decision bindings govern the final lock. + +## Forbidden transitions + +- Approving a rejected candidate or rejecting an approved candidate. +- Approving a second candidate in the same resolution run. +- Locking an unapproved candidate or a candidate different from the approved + candidate. +- Any new or conflicting decision after lock. +- Reusing the same candidate decision with a different status, reviewer, note, + or approval mode. +- Committing against an explicitly supplied stale trace revision. + +All conflicts fail closed with stable domain errors. The administration API maps +state conflicts and stale revisions to stable HTTP 409 reason codes and never +returns internal exception text. + +## Integrity bindings + +Every decision remains bound to the exact candidate digest, recommendation +digest and revision, database anchor, semantic registry anchor, policy anchor, +verified reviewer identity, and the append-only trace prefix created by that +decision. + +A later legal decision may extend the trace but cannot invalidate an earlier +approval. Locking verifies the approval's recorded trace revision and chain hash +against the corresponding immutable prefix of the current trace. Rewritten, +missing, or non-prefix history fails closed. + +## Concurrency and retry + +The store is the atomic authority for decision and lock commits. + +- Compare-and-set checks cover the recommendation digest and trace revision. +- An explicit expected revision is strict: a mismatch raises + `STALE_REVIEW_REVISION` and is not retried. +- Without an explicit revision, the engine may retry a bounded number of times + after a concurrent legal trace append. +- Concurrent exact duplicate operations converge on one stored decision or lock. +- Concurrent conflicting approvals have one winner; all other attempts fail + closed. Double lock is impossible. +- Reconstructing the engine over the same store preserves state and permits the + next legal transition. + +## Administration API identity + +Review mutations exist only on the isolated administration application. Request +bodies do not accept a reviewer identity. The reviewer is always the +`actor_id` supplied by the verified authorization context. The public +`create_app` surface remains read/resolve-only and unchanged. diff --git a/src/a1_factor_engine/__init__.py b/src/a1_factor_engine/__init__.py index 32d0b4a..42171be 100644 --- a/src/a1_factor_engine/__init__.py +++ b/src/a1_factor_engine/__init__.py @@ -20,7 +20,12 @@ interpret_process_emission_observation, ) from .engine import A1FactorResolutionEngine -from .integrity import CatalogIntegrityError, PersistenceIntegrityError +from .integrity import ( + CatalogIntegrityError, + PersistenceIntegrityError, + ReviewStateConflictError, + StaleReviewRevisionError, +) from .material_registry import ( DEFAULT_MATERIAL_REGISTRY, FormRule, @@ -180,6 +185,8 @@ "LockedResolutionEvidenceSnapshot", "CatalogIntegrityError", "PersistenceIntegrityError", + "ReviewStateConflictError", + "StaleReviewRevisionError", "LinkAttempt", "LinkOutcome", "LinkStrategy", diff --git a/src/a1_factor_engine/adapters.py b/src/a1_factor_engine/adapters.py index 38db7fc..2fa39df 100644 --- a/src/a1_factor_engine/adapters.py +++ b/src/a1_factor_engine/adapters.py @@ -21,6 +21,9 @@ from .integrity import ( CATALOG_SCHEMA_VERSION, CatalogIntegrityError, + PersistenceIntegrityError, + ReviewStateConflictError, + StaleReviewRevisionError, catalog_content_sha256, stable_sha256, verify_digest, @@ -28,6 +31,7 @@ from .matching import normalize_text from .material_registry import DEFAULT_MATERIAL_REGISTRY, MaterialSemanticRegistryPort from .models import ( + ApprovalMode, ApprovalRecord, ApprovalStatus, DatabaseVersionAnchor, @@ -38,6 +42,7 @@ LinkOutcome, LinkStrategy, LockedResolution, + LockedResolutionEvidenceSnapshot, MaterialCategory, MaterialClass, MaterialInterpretation, @@ -871,6 +876,24 @@ async def save_trace(self, trace: ResolutionTrace) -> None: raise ValueError("trace updates require an existing atomic resolution run") trace.verify_hash_chain() current = self.traces[trace.request_id] + trace_envelope = ( + trace.trace_id, + trace.request_id, + trace.request_fingerprint, + trace.raw_request_fingerprint, + trace.normalized_business_fingerprint, + trace.database_anchor, + ) + current_envelope = ( + current.trace_id, + current.request_id, + current.request_fingerprint, + current.raw_request_fingerprint, + current.normalized_business_fingerprint, + current.database_anchor, + ) + if trace_envelope != current_envelope: + raise PersistenceIntegrityError("trace envelope is immutable") if trace.revision < current.revision: raise ValueError("trace revision cannot move backwards") current_hashes = tuple(item.entry_hash for item in current.entries) @@ -895,42 +918,160 @@ async def save_approval( *, expected_recommendation_sha256: str, expected_trace_revision: int, - ) -> None: + ) -> ApprovalRecord: key = (approval.request_id, approval.candidate_id) async with self._approval_lock: - if approval.request_id in self.locked: - raise ValueError("locked resolution cannot be changed") recommendation = self.recommendations.get(approval.request_id) current_trace = self.traces.get(approval.request_id) if recommendation is None or current_trace is None: - raise ValueError("approval requires an existing resolution run") - if recommendation.content_sha256 != expected_recommendation_sha256: - raise ValueError("recommendation changed before approval commit") + raise ReviewStateConflictError("approval requires an existing resolution run") if current_trace.revision != expected_trace_revision: - raise ValueError("trace revision changed before approval commit") + raise StaleReviewRevisionError("trace revision changed before approval commit") + existing = self.approvals.get(key) + if existing is not None: + if existing.matches_decision( + status=approval.status, + reviewer=approval.reviewer, + note=approval.note, + mode=approval.mode, + ): + stable_existing_bindings = ( + existing.candidate_content_sha256, + existing.recommendation_content_sha256, + existing.recommendation_revision, + existing.database_anchor_sha256, + existing.registry_anchor_sha256, + existing.policy_anchor_sha256, + ) + stable_incoming_bindings = ( + approval.candidate_content_sha256, + approval.recommendation_content_sha256, + approval.recommendation_revision, + approval.database_anchor_sha256, + approval.registry_anchor_sha256, + approval.policy_anchor_sha256, + ) + if stable_existing_bindings == stable_incoming_bindings: + return existing + raise PersistenceIntegrityError( + "replayed decision bindings do not match committed state" + ) + raise ReviewStateConflictError( + "candidate already has a different terminal human decision" + ) + if approval.request_id in self.locked: + raise ReviewStateConflictError("locked resolution cannot be changed") + if recommendation.content_sha256 != expected_recommendation_sha256: + raise ReviewStateConflictError("recommendation changed before approval commit") trace.verify_hash_chain() + trace_envelope = ( + trace.trace_id, + trace.request_id, + trace.request_fingerprint, + trace.raw_request_fingerprint, + trace.normalized_business_fingerprint, + trace.database_anchor, + ) + current_envelope = ( + current_trace.trace_id, + current_trace.request_id, + current_trace.request_fingerprint, + current_trace.raw_request_fingerprint, + current_trace.normalized_business_fingerprint, + current_trace.database_anchor, + ) + if trace_envelope != current_envelope: + raise PersistenceIntegrityError("approval trace envelope cannot be replaced") if trace.revision != expected_trace_revision + 1: - raise ValueError("approval commit must append exactly one trace event") + raise PersistenceIntegrityError( + "approval commit must append exactly one trace event" + ) if tuple(item.entry_hash for item in trace.entries[:-1]) != tuple( item.entry_hash for item in current_trace.entries ): - raise ValueError("approval trace does not extend the stored trace") - if key in self.approvals: - raise ValueError("candidate already has a terminal human decision") + raise PersistenceIntegrityError( + "approval trace does not extend the stored trace" + ) + if approval.status not in {ApprovalStatus.APPROVED, ApprovalStatus.REJECTED}: + raise ReviewStateConflictError( + "only approved or rejected terminal decisions can be persisted" + ) + if ( + approval.status == ApprovalStatus.REJECTED + and approval.mode != ApprovalMode.STANDARD + ): + raise ReviewStateConflictError( + "rejected decisions require standard approval mode" + ) if approval.status == ApprovalStatus.APPROVED and any( item.request_id == approval.request_id and item.status == ApprovalStatus.APPROVED for item in self.approvals.values() ): - raise ValueError("request already has an approved candidate") + raise ReviewStateConflictError("request already has an approved candidate") if not approval.is_integrity_bound: - raise ValueError("legacy approval without integrity digests cannot be persisted") + raise PersistenceIntegrityError( + "legacy approval without integrity digests cannot be persisted" + ) + expected_message = ( + "candidate approved" + if approval.status == ApprovalStatus.APPROVED + else "candidate rejected" + ) + expected_details: dict[str, object] = { + "candidate_id": approval.candidate_id, + "reviewer": approval.reviewer, + "note": approval.note, + } + if approval.status == ApprovalStatus.APPROVED: + expected_details["approval_mode"] = approval.mode.value + event = trace.entries[-1] + if ( + event.stage != "human_approval" + or event.message != expected_message + or dict(event.details) != expected_details + or approval.reviewer_identity != approval.reviewer + ): + raise PersistenceIntegrityError( + "approval decision does not match its trace event or verified reviewer" + ) + candidate = next(( + item for item in ( + *recommendation.candidates, *recommendation.reviewable_candidates + ) + if item.candidate_id == approval.candidate_id + ), None) + expected_bindings = ( + candidate.content_sha256 if candidate is not None else None, + recommendation.content_sha256, + recommendation.revision, + recommendation.database_anchor_sha256, + recommendation.registry_anchor_sha256, + recommendation.policy_anchor_sha256, + trace.revision, + trace.chain_sha256, + ) + observed_bindings = ( + approval.candidate_content_sha256, + approval.recommendation_content_sha256, + approval.recommendation_revision, + approval.database_anchor_sha256, + approval.registry_anchor_sha256, + approval.policy_anchor_sha256, + approval.trace_revision, + approval.trace_chain_sha256, + ) + if candidate is None or observed_bindings != expected_bindings: + raise PersistenceIntegrityError( + "approval bindings do not match the committed candidate, recommendation, anchors or trace" + ) self.approvals[key] = approval stored_trace = trace.clone() self.traces[trace.request_id] = stored_trace self.recommendations[trace.request_id] = replace( recommendation, trace=stored_trace ) + return approval async def get_approval(self, request_id: str, candidate_id: str) -> ApprovalRecord | None: return self.approvals.get((request_id, candidate_id)) @@ -942,34 +1083,104 @@ async def save_locked( *, expected_recommendation_sha256: str, expected_trace_revision: int, - ) -> None: + ) -> LockedResolution: async with self._approval_lock: - if locked.request_id in self.locked: - raise ValueError("resolution is already locked and immutable") recommendation = self.recommendations.get(locked.request_id) current_trace = self.traces.get(locked.request_id) if recommendation is None or current_trace is None: - raise ValueError("lock requires an existing resolution run") - if recommendation.content_sha256 != expected_recommendation_sha256: - raise ValueError("recommendation changed before lock commit") + raise ReviewStateConflictError("lock requires an existing resolution run") if current_trace.revision != expected_trace_revision: - raise ValueError("trace revision changed before lock commit") + raise StaleReviewRevisionError("trace revision changed before lock commit") + existing = self.locked.get(locked.request_id) + if existing is not None: + if existing.candidate.candidate_id == locked.candidate.candidate_id: + return existing + raise ReviewStateConflictError("resolution is already locked and immutable") + if recommendation.content_sha256 != expected_recommendation_sha256: + raise ReviewStateConflictError("recommendation changed before lock commit") trace.verify_hash_chain() + trace_envelope = ( + trace.trace_id, + trace.request_id, + trace.request_fingerprint, + trace.raw_request_fingerprint, + trace.normalized_business_fingerprint, + trace.database_anchor, + ) + current_envelope = ( + current_trace.trace_id, + current_trace.request_id, + current_trace.request_fingerprint, + current_trace.raw_request_fingerprint, + current_trace.normalized_business_fingerprint, + current_trace.database_anchor, + ) + if trace_envelope != current_envelope: + raise PersistenceIntegrityError("lock trace envelope cannot be replaced") if trace.revision != expected_trace_revision + 1: - raise ValueError("lock commit must append exactly one trace event") + raise PersistenceIntegrityError("lock commit must append exactly one trace event") if tuple(item.entry_hash for item in trace.entries[:-1]) != tuple( item.entry_hash for item in current_trace.entries ): - raise ValueError("lock trace does not extend the stored trace") + raise PersistenceIntegrityError("lock trace does not extend the stored trace") approval = self.approvals.get((locked.request_id, locked.candidate.candidate_id)) if approval is None or approval.content_sha256 != locked.approval_content_sha256: - raise ValueError("approval changed before lock commit") + raise PersistenceIntegrityError("approval changed before lock commit") + if locked.approval != replace(approval, status=ApprovalStatus.LOCKED): + raise PersistenceIntegrityError( + "lock embeds an approval different from the canonical stored approval" + ) + canonical_candidate = next(( + item for item in ( + *recommendation.candidates, *recommendation.reviewable_candidates + ) + if item.candidate_id == locked.candidate.candidate_id + ), None) + if ( + canonical_candidate is None + or locked.candidate != canonical_candidate + or locked.candidate.content_sha256 != approval.candidate_content_sha256 + ): + raise PersistenceIntegrityError( + "lock candidate differs from the approved recommendation candidate" + ) + expected_snapshot = LockedResolutionEvidenceSnapshot.from_trace( + trace, + registry_anchor_sha256=recommendation.registry_anchor_sha256 or "", + policy_anchor_sha256=recommendation.policy_anchor_sha256 or "", + ) + expected_lock_details = { + "candidate_id": locked.candidate.candidate_id, + "reviewer": locked.reviewer, + "locked_result_is_immutable": True, + "trace_remains_appendable": True, + "result_tier": locked.candidate.result_tier.value, + "approval_mode": approval.mode.value, + "accepted_assumptions": locked.candidate.assumptions, + "override_reason": ( + approval.note + if approval.mode == ApprovalMode.REFERENCE_OVERRIDE + else None + ), + "unresolved_warnings": locked.candidate.warnings, + } + lock_event = trace.entries[-1] + if ( + locked.candidate_content_sha256 != locked.candidate.content_sha256 + or locked.recommendation_content_sha256 != recommendation.content_sha256 + or locked.evidence_snapshot != expected_snapshot + or lock_event.stage != "lock" + or lock_event.message != "approved factor result locked" + or dict(lock_event.details) != expected_lock_details + ): + raise PersistenceIntegrityError("lock bindings do not match committed state") self.locked[locked.request_id] = locked stored_trace = trace.clone() self.traces[trace.request_id] = stored_trace self.recommendations[trace.request_id] = replace( recommendation, trace=stored_trace ) + return locked async def get_locked(self, request_id: str) -> LockedResolution | None: return self.locked.get(request_id) diff --git a/src/a1_factor_engine/api.py b/src/a1_factor_engine/api.py index a4247c9..0ec550d 100644 --- a/src/a1_factor_engine/api.py +++ b/src/a1_factor_engine/api.py @@ -17,6 +17,11 @@ from uuid import uuid4 from .engine import A1FactorResolutionEngine +from .integrity import ( + PersistenceIntegrityError, + ReviewStateConflictError, + StaleReviewRevisionError, +) from .models import ResolutionRequest, resolution_request_fingerprint from .operability import ( API_CONTRACT_VERSION, @@ -43,6 +48,9 @@ BENCHMARK_DATASET_REJECTED = "BENCHMARK_DATASET_REJECTED" RESOLUTION_SCOPE_CONFLICT = "RESOLUTION_SCOPE_CONFLICT" RESOLUTION_PAYLOAD_CONFLICT = "RESOLUTION_PAYLOAD_CONFLICT" +REVIEW_STATE_CONFLICT = "REVIEW_STATE_CONFLICT" +REVIEW_INTEGRITY_CONFLICT = "REVIEW_INTEGRITY_CONFLICT" +STALE_REVIEW_REVISION = "STALE_REVIEW_REVISION" MAX_BENCHMARK_BYTES = 2_000_000 MAX_BENCHMARK_RUNS = 64 MAX_BENCHMARK_CACHE_BYTES = 8_000_000 @@ -172,7 +180,7 @@ def _engine_is_configured(engine: Any, *, explicitly_supplied: bool) -> bool: return not isinstance(graph.local_retrieval, NullFactorRepository) or bool(graph.external_connectors) -def _install_http_contract(app: Any) -> None: +def _install_http_contract(app: Any, *, admin_review_routes: bool = False) -> None: from fastapi import Request from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse @@ -192,9 +200,17 @@ async def correlation_contract(request: Request, call_next: Callable[..., Any]): supplied = request.headers.get("x-request-id") or request.headers.get(CORRELATION_ID_HEADER) request.state.correlation_id = safe_request_id(supplied) media_type = request.headers.get("content-type", "").split(";", 1)[0].strip().casefold() + review_json_route = ( + admin_review_routes + and request.url.path.startswith("/api/v1/resolutions/") + and request.url.path.rsplit("/", 1)[-1] in {"decisions", "locks"} + ) if ( request.method == "POST" - and request.url.path in {"/api/v1/resolve", "/api/v1/debug/resolve"} + and ( + request.url.path in {"/api/v1/resolve", "/api/v1/debug/resolve"} + or review_json_route + ) and media_type != "application/json" and not media_type.endswith("+json") ): @@ -600,6 +616,20 @@ def create_admin_app( # Route annotations are postponed; expose the lazily imported type for # FastAPI/Pydantic resolution without making FastAPI a core dependency. globals()["Request"] = Request + from .api_contracts import ( + PublicErrorEnvelopeDTO, + ReviewDecisionRequestDTO, + ReviewDecisionResponseDTO, + ReviewLockRequestDTO, + ReviewLockResponseDTO, + review_decision_dto, + review_lock_dto, + ) + + globals()["ReviewDecisionRequestDTO"] = ReviewDecisionRequestDTO + globals()["ReviewDecisionResponseDTO"] = ReviewDecisionResponseDTO + globals()["ReviewLockRequestDTO"] = ReviewLockRequestDTO + globals()["ReviewLockResponseDTO"] = ReviewLockResponseDTO resolver = engine or _unconfigured_engine() engine_configured = _engine_is_configured(resolver, explicitly_supplied=engine is not None) health = _connector_health_for(resolver, connector_health) @@ -613,7 +643,7 @@ def create_admin_app( app.state.benchmark_runner = benchmark_runner app.state.benchmark_runs = runs app.state.resolution_fingerprints = resolution_fingerprints - _install_http_contract(app) + _install_http_contract(app, admin_review_routes=True) async def require(request: Request, permission: str) -> AuthorizationContext: try: @@ -653,6 +683,115 @@ def require_resolution_owner(context: AuthorizationContext, request_id: str) -> if resolution_owners.get(request_id) != (context.tenant_id, context.project_id): raise HTTPException(status_code=404, detail="resolution not found") + review_conflict_responses: dict[int | str, dict[str, Any]] = { + 409: _documented_error( + "Review state or revision conflict", + REVIEW_STATE_CONFLICT, + "review operation conflicts with committed state", + model=PublicErrorEnvelopeDTO, + ), + 415: _documented_error( + "JSON media type required", + UNSUPPORTED_MEDIA_TYPE, + "application/json is required", + model=PublicErrorEnvelopeDTO, + ), + 422: _documented_error( + "JSON request validation failed", + REQUEST_VALIDATION_FAILED, + "request validation failed", + model=PublicErrorEnvelopeDTO, + ), + } + + def raise_review_error(exc: Exception) -> None: + if isinstance(exc, StaleReviewRevisionError): + reason = STALE_REVIEW_REVISION + message = "review trace revision is stale" + elif isinstance(exc, PersistenceIntegrityError): + reason = REVIEW_INTEGRITY_CONFLICT + message = "review integrity validation failed" + else: + reason = REVIEW_STATE_CONFLICT + message = "review operation conflicts with committed state" + raise HTTPException( + status_code=409, + detail={"reason_code": reason, "message": message}, + ) from exc + + @app.post( + "/api/v1/resolutions/{request_id}/decisions", + response_model=ReviewDecisionResponseDTO, + responses=review_conflict_responses, + openapi_extra={"x-cfr-reason-codes": [ + REVIEW_STATE_CONFLICT, + REVIEW_INTEGRITY_CONFLICT, + STALE_REVIEW_REVISION, + ADMIN_AUTHORIZATION_REQUIRED, + ]}, + ) + async def create_review_decision( + request: Request, + request_id: str, + payload: ReviewDecisionRequestDTO, + ) -> dict[str, Any]: + context = await require(request, "review:write") + require_resolution_owner(context, request_id) + try: + if payload.decision == "approve": + decision = await resolver.approve( + request_id, + payload.candidate_id, + context.actor_id, + payload.note, + payload.mode, + expected_trace_revision=payload.expected_trace_revision, + ) + else: + decision = await resolver.reject( + request_id, + payload.candidate_id, + context.actor_id, + payload.note, + expected_trace_revision=payload.expected_trace_revision, + ) + except KeyError as exc: + raise HTTPException(status_code=404, detail="review target not found") from exc + except (PersistenceIntegrityError, ReviewStateConflictError, ValueError) as exc: + raise_review_error(exc) + return review_decision_dto(decision).model_dump(mode="json") + + @app.post( + "/api/v1/resolutions/{request_id}/locks", + response_model=ReviewLockResponseDTO, + responses=review_conflict_responses, + openapi_extra={"x-cfr-reason-codes": [ + REVIEW_STATE_CONFLICT, + REVIEW_INTEGRITY_CONFLICT, + STALE_REVIEW_REVISION, + ADMIN_AUTHORIZATION_REQUIRED, + ]}, + ) + async def lock_reviewed_resolution( + request: Request, + request_id: str, + payload: ReviewLockRequestDTO, + ) -> dict[str, Any]: + context = await require(request, "review:lock") + require_resolution_owner(context, request_id) + try: + locked = await resolver.lock( + request_id, + payload.candidate_id, + context.actor_id, + expected_trace_revision=payload.expected_trace_revision, + ) + except KeyError as exc: + raise HTTPException(status_code=404, detail="review target not found") from exc + except (PersistenceIntegrityError, ReviewStateConflictError, ValueError) as exc: + raise_review_error(exc) + return review_lock_dto(locked).model_dump(mode="json") + @app.post("/api/v1/debug/resolve") async def resolve_debug(request: Request, payload: dict[str, Any] = Body(...)) -> dict[str, Any]: context = await require(request, "resolve:debug") @@ -824,6 +963,9 @@ async def compare_benchmark_runs(request: Request, base: str, candidate: str) -> "INVALID_RESOLUTION_REQUEST", "RESOLUTION_PAYLOAD_CONFLICT", "RESOLUTION_SCOPE_CONFLICT", + "REVIEW_INTEGRITY_CONFLICT", + "REVIEW_STATE_CONFLICT", + "STALE_REVIEW_REVISION", "create_admin_app", "create_app", ] diff --git a/src/a1_factor_engine/api_contracts.py b/src/a1_factor_engine/api_contracts.py index db813bb..fd791d5 100644 --- a/src/a1_factor_engine/api_contracts.py +++ b/src/a1_factor_engine/api_contracts.py @@ -9,12 +9,14 @@ from __future__ import annotations from collections.abc import Mapping, Sequence -from typing import Annotated, Any +from typing import Annotated, Any, Literal from pydantic import BaseModel, ConfigDict, Field, StringConstraints, model_validator from pydantic.json_schema import SkipJsonSchema from .models import ( + ApprovalMode, + ApprovalStatus, CandidateOrigin, FactorKind, FactorSourceType, @@ -37,6 +39,7 @@ ] StrictYear = Annotated[int, Field(strict=True, gt=0, lt=3000)] StrictTopK = Annotated[int, Field(strict=True, ge=1, le=50)] +StrictRevision = Annotated[int, Field(strict=True, ge=0)] OptionalNonEmptyText = NonEmptyText | SkipJsonSchema[None] OptionalStrictYear = StrictYear | SkipJsonSchema[None] @@ -89,6 +92,55 @@ def to_domain_mapping(self) -> dict[str, object]: return self.model_dump(mode="python", exclude_none=True) +class ReviewDecisionRequestDTO(_StrictRequestModel): + """Admin-only decision command; reviewer identity is never caller supplied.""" + + candidate_id: NonEmptyText + decision: Literal["approve", "reject"] + note: Annotated[str, StringConstraints(strict=True)] = "" + mode: ApprovalMode = ApprovalMode.STANDARD + expected_trace_revision: StrictRevision | SkipJsonSchema[None] = None + + @model_validator(mode="after") + def reject_mode_must_be_standard(self) -> ReviewDecisionRequestDTO: + if self.decision == "reject" and self.mode != ApprovalMode.STANDARD: + raise ValueError("reject decisions only support standard mode") + return self + + +class ReviewLockRequestDTO(_StrictRequestModel): + """Admin-only immutable lock command.""" + + candidate_id: NonEmptyText + expected_trace_revision: StrictRevision | SkipJsonSchema[None] = None + + +class ReviewDecisionResponseDTO(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + request_id: str + candidate_id: str + status: ApprovalStatus + reviewer_identity: str + note: str + mode: ApprovalMode + trace_revision: int + trace_chain_sha256: str + content_sha256: str + + +class ReviewLockResponseDTO(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + request_id: str + candidate_id: str + status: ApprovalStatus + reviewer_identity: str + trace_revision: int + trace_chain_sha256: str + content_sha256: str + + class PublicErrorDetailDTO(BaseModel): model_config = ConfigDict(extra="forbid", frozen=True) @@ -308,11 +360,45 @@ def public_recommendation_dto( ) +def review_decision_dto(decision: Any) -> ReviewDecisionResponseDTO: + return ReviewDecisionResponseDTO( + request_id=_text(_field(decision, "request_id")), + candidate_id=_text(_field(decision, "candidate_id")), + status=_field(decision, "status"), + reviewer_identity=_text(_field(decision, "reviewer_identity")), + note=_text(_field(decision, "note", "")), + mode=_field(decision, "mode"), + trace_revision=_field(decision, "trace_revision"), + trace_chain_sha256=_text(_field(decision, "trace_chain_sha256")), + content_sha256=_text(_field(decision, "content_sha256")), + ) + + +def review_lock_dto(locked: Any) -> ReviewLockResponseDTO: + snapshot = _field(locked, "evidence_snapshot") + approval = _field(locked, "approval") + return ReviewLockResponseDTO( + request_id=_text(_field(locked, "request_id")), + candidate_id=_text(_field(_field(locked, "candidate"), "candidate_id")), + status=_field(approval, "status"), + reviewer_identity=_text(_field(locked, "reviewer")), + trace_revision=_field(snapshot, "trace_revision"), + trace_chain_sha256=_text(_field(snapshot, "trace_chain_sha256")), + content_sha256=_text(_field(locked, "content_sha256")), + ) + + __all__ = [ "PublicErrorEnvelopeDTO", "PublicReadinessErrorEnvelopeDTO", "PublicRecommendationDTO", + "ReviewDecisionRequestDTO", + "ReviewDecisionResponseDTO", + "ReviewLockRequestDTO", + "ReviewLockResponseDTO", "ResolutionRequestDTO", "UnitConversionEvidenceDTO", "public_recommendation_dto", + "review_decision_dto", + "review_lock_dto", ] diff --git a/src/a1_factor_engine/engine.py b/src/a1_factor_engine/engine.py index 982173b..76d37ed 100644 --- a/src/a1_factor_engine/engine.py +++ b/src/a1_factor_engine/engine.py @@ -16,6 +16,11 @@ ) from .derived_factor import expected_total_emissions from .graph import GraphState, Router, Stage, candidate_hard_rejection_reasons +from .integrity import ( + PersistenceIntegrityError, + ReviewStateConflictError, + StaleReviewRevisionError, +) from .material_registry import ( DEFAULT_MATERIAL_REGISTRY, MaterialRuleSuggestionPort, @@ -513,6 +518,7 @@ async def trace(self, request_id: str) -> ResolutionTrace | None: async def approve( self, request_id: str, candidate_id: str, reviewer: str, note: str = "", mode: ApprovalMode | str = ApprovalMode.STANDARD, + *, expected_trace_revision: int | None = None, ) -> ApprovalRecord: recommendation = await self.store.get_recommendation(request_id) if recommendation is None: @@ -564,30 +570,25 @@ async def approve( raise ValueError("REFERENCE_ONLY candidates require reference_override mode and a non-empty reason") elif candidate.result_tier == ResultTier.USABLE_WITH_ASSUMPTIONS and mode == ApprovalMode.STANDARD: raise ValueError("USABLE_WITH_ASSUMPTIONS candidates require assumption_acceptance mode") - existing_approval = await self.store.get_approval(request_id, candidate_id) - if existing_approval is not None and existing_approval.status == ApprovalStatus.REJECTED: - raise ValueError("rejected candidate cannot be approved in the same resolution run") - trace = await self.store.get_trace(request_id) - if trace is None: - raise KeyError(f"trace not found: {request_id}") - expected_trace_revision = trace.revision - updated_trace = trace.clone() - updated_trace.append("human_approval", "candidate approved", { - "candidate_id": candidate_id, "reviewer": reviewer, "note": note, "approval_mode": mode.value, - }) - approval = self._bound_approval( - recommendation, candidate, updated_trace, reviewer, ApprovalStatus.APPROVED, - note=note, mode=mode, - ) - await self.store.save_approval( - approval, - updated_trace, - expected_recommendation_sha256=recommendation.content_sha256, + return await self._commit_decision( + recommendation, + candidate, + reviewer, + ApprovalStatus.APPROVED, + note=note, + mode=mode, expected_trace_revision=expected_trace_revision, ) - return approval - async def reject(self, request_id: str, candidate_id: str, reviewer: str, note: str = "") -> ApprovalRecord: + async def reject( + self, + request_id: str, + candidate_id: str, + reviewer: str, + note: str = "", + *, + expected_trace_revision: int | None = None, + ) -> ApprovalRecord: recommendation = await self.store.get_recommendation(request_id) if recommendation is None: raise KeyError(f"unknown request: {request_id}") @@ -599,35 +600,38 @@ async def reject(self, request_id: str, candidate_id: str, reviewer: str, note: ), None) if candidate is None: raise KeyError(f"candidate not found: {candidate_id}") - trace = await self.store.get_trace(request_id) - if trace is None: - raise KeyError(f"trace not found: {request_id}") - expected_trace_revision = trace.revision - updated_trace = trace.clone() - updated_trace.append("human_approval", "candidate rejected", { - "candidate_id": candidate_id, "reviewer": reviewer, "note": note, - }) - approval = self._bound_approval( - recommendation, candidate, updated_trace, reviewer, ApprovalStatus.REJECTED, + return await self._commit_decision( + recommendation, + candidate, + reviewer, + ApprovalStatus.REJECTED, note=note, - ) - await self.store.save_approval( - approval, - updated_trace, - expected_recommendation_sha256=recommendation.content_sha256, + mode=ApprovalMode.STANDARD, expected_trace_revision=expected_trace_revision, ) - return approval - async def lock(self, request_id: str, candidate_id: str, reviewer: str) -> LockedResolution: + async def lock( + self, + request_id: str, + candidate_id: str, + reviewer: str, + *, + expected_trace_revision: int | None = None, + ) -> LockedResolution: + if expected_trace_revision is not None: + current_trace = await self.store.get_trace(request_id) + if current_trace is None: + raise KeyError(f"trace not found: {request_id}") + if current_trace.revision != expected_trace_revision: + raise StaleReviewRevisionError("trace revision changed before lock") existing = await self.store.get_locked(request_id) if existing is not None: - if existing.candidate.candidate_id != candidate_id: - raise ValueError("resolution is already locked and immutable") - return existing + if existing.candidate.candidate_id == candidate_id: + return existing + raise ReviewStateConflictError("resolution is already locked and immutable") approval = await self.store.get_approval(request_id, candidate_id) if approval is None or approval.status != ApprovalStatus.APPROVED: - raise ValueError("candidate must be approved before locking") + raise ReviewStateConflictError("candidate must be approved before locking") recommendation = await self.store.get_recommendation(request_id) if recommendation is None: raise KeyError(f"unknown request: {request_id}") @@ -639,9 +643,6 @@ async def lock(self, request_id: str, candidate_id: str, reviewer: str) -> Locke ), None) if candidate is None: raise KeyError(f"candidate not found: {candidate_id}") - trace = await self.store.get_trace(request_id) - if trace is None: - raise KeyError(f"trace not found: {request_id}") incomplete = tuple( item for item in candidate.limitations if item.startswith("formal_admission_incomplete:") @@ -678,8 +679,6 @@ async def lock(self, request_id: str, candidate_id: str, reviewer: str) -> Locke recommendation.database_anchor_sha256, recommendation.registry_anchor_sha256, recommendation.policy_anchor_sha256, - trace.revision, - trace.chain_sha256, ) observed_bindings = ( approval.candidate_content_sha256, @@ -688,47 +687,69 @@ async def lock(self, request_id: str, candidate_id: str, reviewer: str) -> Locke approval.database_anchor_sha256, approval.registry_anchor_sha256, approval.policy_anchor_sha256, - approval.trace_revision, - approval.trace_chain_sha256, ) if observed_bindings != expected_bindings: - raise ValueError("approval binding no longer matches candidate, recommendation, anchors or trace") - expected_trace_revision = trace.revision - updated_trace = trace.clone() - updated_trace.append("lock", "approved factor result locked", { - "candidate_id": candidate_id, - "reviewer": reviewer, - "locked_result_is_immutable": True, - "trace_remains_appendable": True, - "result_tier": candidate.result_tier.value, - "approval_mode": approval.mode.value, - "accepted_assumptions": candidate.assumptions, - "override_reason": approval.note if approval.mode == ApprovalMode.REFERENCE_OVERRIDE else None, - "unresolved_warnings": candidate.warnings, - }) - snapshot = LockedResolutionEvidenceSnapshot.from_trace( - updated_trace, - registry_anchor_sha256=recommendation.registry_anchor_sha256 or "", - policy_anchor_sha256=recommendation.policy_anchor_sha256 or "", - ) - locked_approval = replace(approval, status=ApprovalStatus.LOCKED) - locked = LockedResolution( - request_id=request_id, - candidate=candidate, - reviewer=reviewer, - approval=locked_approval, - evidence_snapshot=snapshot, - candidate_content_sha256=candidate.content_sha256, - recommendation_content_sha256=recommendation.content_sha256, - approval_content_sha256=approval.content_sha256, - ) - await self.store.save_locked( - locked, - updated_trace, - expected_recommendation_sha256=recommendation.content_sha256, - expected_trace_revision=expected_trace_revision, - ) - return locked + raise PersistenceIntegrityError( + "approval binding no longer matches candidate, recommendation or anchors" + ) + for attempt in range(8): + trace = await self.store.get_trace(request_id) + if trace is None: + raise KeyError(f"trace not found: {request_id}") + if ( + expected_trace_revision is not None + and trace.revision != expected_trace_revision + ): + raise StaleReviewRevisionError("trace revision changed before lock") + if ( + approval.trace_revision is None + or approval.trace_revision < 1 + or approval.trace_revision > trace.revision + or approval.trace_chain_sha256 + != trace.chain_sha256_at_revision(approval.trace_revision) + ): + raise PersistenceIntegrityError( + "approval trace binding is not an immutable prefix of the current trace" + ) + commit_revision = trace.revision + updated_trace = trace.clone() + updated_trace.append("lock", "approved factor result locked", { + "candidate_id": candidate_id, + "reviewer": reviewer, + "locked_result_is_immutable": True, + "trace_remains_appendable": True, + "result_tier": candidate.result_tier.value, + "approval_mode": approval.mode.value, + "accepted_assumptions": candidate.assumptions, + "override_reason": approval.note if approval.mode == ApprovalMode.REFERENCE_OVERRIDE else None, + "unresolved_warnings": candidate.warnings, + }) + snapshot = LockedResolutionEvidenceSnapshot.from_trace( + updated_trace, + registry_anchor_sha256=recommendation.registry_anchor_sha256 or "", + policy_anchor_sha256=recommendation.policy_anchor_sha256 or "", + ) + locked = LockedResolution( + request_id=request_id, + candidate=candidate, + reviewer=reviewer, + approval=replace(approval, status=ApprovalStatus.LOCKED), + evidence_snapshot=snapshot, + candidate_content_sha256=candidate.content_sha256, + recommendation_content_sha256=recommendation.content_sha256, + approval_content_sha256=approval.content_sha256, + ) + try: + return await self.store.save_locked( + locked, + updated_trace, + expected_recommendation_sha256=recommendation.content_sha256, + expected_trace_revision=commit_revision, + ) + except StaleReviewRevisionError: + if expected_trace_revision is not None or attempt == 7: + raise + raise StaleReviewRevisionError("lock could not commit after concurrent updates") async def locked(self, request_id: str) -> LockedResolution | None: return await self.store.get_locked(request_id) @@ -811,6 +832,82 @@ def _bind_recommendation( policy_anchor_sha256=(policy or self.graph.deployment_policy).anchor_sha256, ) + async def _commit_decision( + self, + recommendation: Recommendation, + candidate: Candidate, + reviewer: str, + status: ApprovalStatus, + *, + note: str, + mode: ApprovalMode, + expected_trace_revision: int | None, + ) -> ApprovalRecord: + if expected_trace_revision is not None: + current_trace = await self.store.get_trace(recommendation.request_id) + if current_trace is None: + raise KeyError(f"trace not found: {recommendation.request_id}") + if current_trace.revision != expected_trace_revision: + raise StaleReviewRevisionError("trace revision changed before decision") + existing = await self.store.get_approval( + recommendation.request_id, candidate.candidate_id + ) + if existing is not None: + if existing.matches_decision( + status=status, reviewer=reviewer, note=note, mode=mode + ): + return existing + if ( + existing.status == ApprovalStatus.REJECTED + and status == ApprovalStatus.APPROVED + ): + raise ReviewStateConflictError( + "rejected candidate cannot be approved in the same resolution run" + ) + raise ReviewStateConflictError( + "candidate already has a different terminal human decision" + ) + for attempt in range(8): + trace = await self.store.get_trace(recommendation.request_id) + if trace is None: + raise KeyError(f"trace not found: {recommendation.request_id}") + if ( + expected_trace_revision is not None + and trace.revision != expected_trace_revision + ): + raise StaleReviewRevisionError("trace revision changed before decision") + commit_revision = trace.revision + updated_trace = trace.clone() + message = "candidate approved" if status == ApprovalStatus.APPROVED else "candidate rejected" + details: dict[str, object] = { + "candidate_id": candidate.candidate_id, + "reviewer": reviewer, + "note": note, + } + if status == ApprovalStatus.APPROVED: + details["approval_mode"] = mode.value + updated_trace.append("human_approval", message, details) + approval = self._bound_approval( + recommendation, + candidate, + updated_trace, + reviewer, + status, + note=note, + mode=mode, + ) + try: + return await self.store.save_approval( + approval, + updated_trace, + expected_recommendation_sha256=recommendation.content_sha256, + expected_trace_revision=commit_revision, + ) + except StaleReviewRevisionError: + if expected_trace_revision is not None or attempt == 7: + raise + raise StaleReviewRevisionError("decision could not commit after concurrent updates") + @staticmethod def _bound_approval( recommendation: Recommendation, diff --git a/src/a1_factor_engine/integrity.py b/src/a1_factor_engine/integrity.py index 94cc3c6..40d951d 100644 --- a/src/a1_factor_engine/integrity.py +++ b/src/a1_factor_engine/integrity.py @@ -32,6 +32,14 @@ class PersistenceIntegrityError(ValueError): """A persisted approval, lock or audit object failed a binding check.""" +class ReviewStateConflictError(ValueError): + """A human-review operation conflicts with an already committed state.""" + + +class StaleReviewRevisionError(ReviewStateConflictError): + """A human-review compare-and-set used an obsolete trace revision.""" + + # Full input schema consumed by HttpCatalogFactorRepository. Unknown fields are # rejected under cfr.catalog/v2 so a result-driving field cannot silently sit # outside the digest contract. Legacy payloads remain readable only through the diff --git a/src/a1_factor_engine/models.py b/src/a1_factor_engine/models.py index c8dea81..f70a3b7 100644 --- a/src/a1_factor_engine/models.py +++ b/src/a1_factor_engine/models.py @@ -1257,7 +1257,14 @@ def verify_hash_chain(self) -> None: @property def chain_sha256(self) -> str: + return self.chain_sha256_at_revision(self.revision) + + def chain_sha256_at_revision(self, revision: int) -> str: + """Hash an immutable prefix using the same contract as the live trace head.""" + self.verify_hash_chain() + if revision < 0 or revision > self.revision: + raise PersistenceIntegrityError("trace prefix revision is outside the stored chain") return stable_sha256({ "schema_version": TRACE_SCHEMA_VERSION, "trace_id": self.trace_id, @@ -1268,8 +1275,10 @@ def chain_sha256(self) -> str: "database_anchor_sha256": ( self.database_anchor.anchor_sha256 if self.database_anchor else None ), - "revision": self.revision, - "entry_hashes": tuple(entry.entry_hash for entry in self.entries), + "revision": revision, + "entry_hashes": tuple( + entry.entry_hash for entry in self.entries[:revision] + ), }) def set_database_anchor(self, anchor: DatabaseVersionAnchor) -> None: @@ -2043,6 +2052,10 @@ class ApprovalRecord: def __post_init__(self) -> None: if not self.request_id.strip() or not self.candidate_id.strip() or not self.reviewer.strip(): raise ValueError("approval requires request_id, candidate_id and reviewer") + if self.reviewer_identity is not None and self.reviewer_identity != self.reviewer: + raise PersistenceIntegrityError( + "approval reviewer identity must match the verified reviewer" + ) if not isinstance(self.mode, ApprovalMode): try: object.__setattr__(self, "mode", ApprovalMode(str(self.mode))) @@ -2084,6 +2097,24 @@ def content_sha256(self) -> str: "trace_chain_sha256": self.trace_chain_sha256, }) + def matches_decision( + self, + *, + status: ApprovalStatus, + reviewer: str, + note: str, + mode: ApprovalMode, + ) -> bool: + """Compare the user-visible command fields, excluding commit-time bindings.""" + + return ( + self.status == status + and self.reviewer == reviewer + and self.reviewer_identity == reviewer + and self.note == note + and self.mode == mode + ) + @dataclass(frozen=True, slots=True) class LockedResolution: diff --git a/src/a1_factor_engine/ports.py b/src/a1_factor_engine/ports.py index 9a522fc..5ed01e3 100644 --- a/src/a1_factor_engine/ports.py +++ b/src/a1_factor_engine/ports.py @@ -105,7 +105,7 @@ async def save_approval( *, expected_recommendation_sha256: str, expected_trace_revision: int, - ) -> None: ... + ) -> ApprovalRecord: ... async def get_approval(self, request_id: str, candidate_id: str) -> ApprovalRecord | None: ... @@ -116,6 +116,6 @@ async def save_locked( *, expected_recommendation_sha256: str, expected_trace_revision: int, - ) -> None: ... + ) -> LockedResolution: ... async def get_locked(self, request_id: str) -> LockedResolution | None: ... diff --git a/tests/test_review_state_machine.py b/tests/test_review_state_machine.py new file mode 100644 index 0000000..95ebb57 --- /dev/null +++ b/tests/test_review_state_machine.py @@ -0,0 +1,544 @@ +from __future__ import annotations + +import asyncio +from dataclasses import replace + +import pytest + +from a1_factor_engine import ( + A1FactorResolutionEngine, + ApprovalMode, + ApprovalRecord, + ApprovalStatus, + FactorKind, + FactorSourceType, + FactorSubjectType, + LockedResolution, + LockedResolutionEvidenceSnapshot, + ResolutionRequest, + ReviewStateConflictError, + SourceRecord, + StaleReviewRevisionError, +) +from a1_factor_engine.adapters import InMemoryFactorRepository, InMemoryResolutionStore +from a1_factor_engine.integrity import PersistenceIntegrityError +from a1_factor_engine.models import TraceEntry + + +def _source(source_id: str, factor_value: float) -> SourceRecord: + return SourceRecord( + source_id=source_id, + source_type=FactorSourceType.LOCAL_DATABASE, + provider="PUBLIC_SYNTHETIC", + locator=f"evidence://{source_id}", + material_name="review material", + factor_value=factor_value, + factor_unit="kgCO2e/kg", + geography="CN", + year=2025, + boundary="cradle-to-gate", + boundary_modules=("A1", "A2", "A3"), + factor_kind=FactorKind.LIFECYCLE_FACTOR, + subject_type=FactorSubjectType.RAW_MATERIAL, + indicator="GWP-total", + declared_product="review material", + source_document_sha256=("ab" if source_id.endswith("a") else "cd") * 32, + ) + + +def _engine() -> A1FactorResolutionEngine: + return A1FactorResolutionEngine( + local_retrieval=InMemoryFactorRepository(( + _source("review-a", 1), + _source("review-b", 2), + )) + ) + + +async def _recommend(engine: A1FactorResolutionEngine, request_id: str): + result = await engine.resolve(ResolutionRequest( + request_id=request_id, + material_name="review material", + quantity=1, + quantity_unit="kg", + geography="CN", + year=2025, + subject_type=FactorSubjectType.RAW_MATERIAL, + top_k=2, + )) + assert len(result.candidates) == 2 + return result, result.candidates[0], result.candidates[1] + + +@pytest.mark.asyncio +async def test_approve_a_then_reject_b_then_lock_a_is_composable() -> None: + engine = _engine() + result, candidate_a, candidate_b = await _recommend(engine, "review-order-red") + + await engine.approve(result.request_id, candidate_a.candidate_id, "reviewer-a") + await engine.reject( + result.request_id, + candidate_b.candidate_id, + "reviewer-b", + "not selected", + ) + locked = await engine.lock(result.request_id, candidate_a.candidate_id, "reviewer-a") + + assert locked.candidate.candidate_id == candidate_a.candidate_id + + +@pytest.mark.asyncio +async def test_reject_b_then_approve_a_then_lock_a_is_composable() -> None: + engine = _engine() + result, candidate_a, candidate_b = await _recommend(engine, "review-reverse") + + await engine.reject(result.request_id, candidate_b.candidate_id, "reviewer-b", "not selected") + await engine.approve(result.request_id, candidate_a.candidate_id, "reviewer-a") + locked = await engine.lock(result.request_id, candidate_a.candidate_id, "reviewer-a") + + assert locked.candidate.candidate_id == candidate_a.candidate_id + + +@pytest.mark.asyncio +async def test_exact_decision_and_lock_replays_are_idempotent() -> None: + engine = _engine() + result, candidate_a, _ = await _recommend(engine, "review-replay") + + before = (await engine.trace(result.request_id)).revision + first = await engine.approve(result.request_id, candidate_a.candidate_id, "reviewer", "selected") + after_first = (await engine.trace(result.request_id)).revision + replay = await engine.approve(result.request_id, candidate_a.candidate_id, "reviewer", "selected") + assert replay is first + assert after_first == before + 1 + assert (await engine.trace(result.request_id)).revision == after_first + + locked = await engine.lock(result.request_id, candidate_a.candidate_id, "locker") + after_lock = (await engine.trace(result.request_id)).revision + retry = await engine.lock(result.request_id, candidate_a.candidate_id, "different-retry-actor") + assert retry is locked + assert retry.reviewer == "locker" + assert (await engine.trace(result.request_id)).revision == after_lock + + +@pytest.mark.asyncio +async def test_same_candidate_conflicts_fail_closed() -> None: + engine = _engine() + result, candidate_a, _ = await _recommend(engine, "review-conflict") + await engine.reject(result.request_id, candidate_a.candidate_id, "reviewer", "bad evidence") + + with pytest.raises(ReviewStateConflictError, match="rejected candidate"): + await engine.approve(result.request_id, candidate_a.candidate_id, "reviewer") + with pytest.raises(ReviewStateConflictError, match="different terminal"): + await engine.reject(result.request_id, candidate_a.candidate_id, "other", "bad evidence") + + +@pytest.mark.asyncio +async def test_explicit_stale_revision_fails_without_trace_mutation() -> None: + engine = _engine() + result, candidate_a, _ = await _recommend(engine, "review-stale") + trace = await engine.trace(result.request_id) + assert trace is not None + stale = trace.revision + await engine.approve( + result.request_id, + candidate_a.candidate_id, + "reviewer", + expected_trace_revision=stale, + ) + committed = (await engine.trace(result.request_id)).revision + + with pytest.raises(StaleReviewRevisionError): + await engine.lock( + result.request_id, + candidate_a.candidate_id, + "reviewer", + expected_trace_revision=stale, + ) + assert (await engine.trace(result.request_id)).revision == committed + + with pytest.raises(StaleReviewRevisionError): + await engine.approve( + result.request_id, + candidate_a.candidate_id, + "reviewer", + expected_trace_revision=stale, + ) + + locked = await engine.lock(result.request_id, candidate_a.candidate_id, "reviewer") + with pytest.raises(StaleReviewRevisionError): + await engine.lock( + result.request_id, + candidate_a.candidate_id, + "reviewer", + expected_trace_revision=committed, + ) + assert await engine.locked(result.request_id) is locked + + +@pytest.mark.asyncio +async def test_engine_reconstruction_over_same_store_recovers_review_state() -> None: + store = InMemoryResolutionStore() + first = A1FactorResolutionEngine( + local_retrieval=InMemoryFactorRepository((_source("review-a", 1), _source("review-b", 2))), + store=store, + ) + result, candidate_a, candidate_b = await _recommend(first, "review-recovery") + await first.approve(result.request_id, candidate_a.candidate_id, "reviewer-a") + + restarted = A1FactorResolutionEngine(store=store) + await restarted.reject(result.request_id, candidate_b.candidate_id, "reviewer-b") + locked = await restarted.lock(result.request_id, candidate_a.candidate_id, "reviewer-a") + assert locked.candidate.candidate_id == candidate_a.candidate_id + + +@pytest.mark.asyncio +async def test_concurrent_legal_decisions_compose_and_competing_approvals_do_not() -> None: + engine = _engine() + result, candidate_a, candidate_b = await _recommend(engine, "review-concurrent-compose") + composed = await asyncio.gather( + engine.approve(result.request_id, candidate_a.candidate_id, "reviewer-a"), + engine.reject(result.request_id, candidate_b.candidate_id, "reviewer-b"), + ) + assert {item.status for item in composed} == {ApprovalStatus.APPROVED, ApprovalStatus.REJECTED} + assert (await engine.lock(result.request_id, candidate_a.candidate_id, "reviewer-a")).candidate == candidate_a + + competing = _engine() + result, candidate_a, candidate_b = await _recommend(competing, "review-concurrent-conflict") + outcomes = await asyncio.gather( + competing.approve(result.request_id, candidate_a.candidate_id, "reviewer-a"), + competing.approve(result.request_id, candidate_b.candidate_id, "reviewer-b"), + return_exceptions=True, + ) + assert sum(isinstance(item, ApprovalRecord) for item in outcomes) == 1 + assert sum(isinstance(item, ReviewStateConflictError) for item in outcomes) == 1 + + +@pytest.mark.asyncio +async def test_concurrent_duplicate_decisions_and_locks_commit_one_event() -> None: + engine = _engine() + result, candidate_a, _ = await _recommend(engine, "review-concurrent-idempotent") + before = (await engine.trace(result.request_id)).revision + decisions = await asyncio.gather(*( + engine.approve(result.request_id, candidate_a.candidate_id, "reviewer", "selected") + for _ in range(8) + )) + assert len({item.content_sha256 for item in decisions}) == 1 + assert (await engine.trace(result.request_id)).revision == before + 1 + + locks = await asyncio.gather(*( + engine.lock(result.request_id, candidate_a.candidate_id, "locker") + for _ in range(8) + )) + assert len({item.content_sha256 for item in locks}) == 1 + assert (await engine.trace(result.request_id)).revision == before + 2 + + +@pytest.mark.asyncio +async def test_mutated_approval_trace_prefix_fails_integrity_check() -> None: + engine = _engine() + result, candidate_a, candidate_b = await _recommend(engine, "review-prefix-integrity") + approval = await engine.approve(result.request_id, candidate_a.candidate_id, "reviewer-a") + await engine.reject(result.request_id, candidate_b.candidate_id, "reviewer-b") + trace = await engine.trace(result.request_id) + assert trace is not None and approval.trace_revision is not None + original = trace.entries[approval.trace_revision - 1] + trace.entries[approval.trace_revision - 1] = TraceEntry( + revision=original.revision, + stage=original.stage, + message="tampered decision", + details=original.details, + at=original.at, + previous_hash=original.previous_hash, + ) + + with pytest.raises(PersistenceIntegrityError): + await engine.lock(result.request_id, candidate_a.candidate_id, "reviewer-a") + + +@pytest.mark.asyncio +async def test_store_rejects_nonterminal_and_forged_replayed_decisions() -> None: + engine = _engine() + result, candidate_a, _ = await _recommend(engine, "review-store-integrity") + trace = await engine.trace(result.request_id) + assert trace is not None + commit_revision = trace.revision + decision_trace = trace.clone() + decision_trace.append("human_approval", "candidate approved", { + "candidate_id": candidate_a.candidate_id, + "reviewer": "reviewer", + "note": "", + "approval_mode": ApprovalMode.STANDARD.value, + }) + approval = engine._bound_approval( + result, + candidate_a, + decision_trace, + "reviewer", + ApprovalStatus.APPROVED, + ) + pending = replace(approval, status=ApprovalStatus.PENDING) + with pytest.raises(ReviewStateConflictError, match="terminal decisions"): + await engine.store.save_approval( + pending, + decision_trace, + expected_recommendation_sha256=result.content_sha256, + expected_trace_revision=commit_revision, + ) + + stored = await engine.store.save_approval( + approval, + decision_trace, + expected_recommendation_sha256=result.content_sha256, + expected_trace_revision=commit_revision, + ) + forged = replace(stored, candidate_content_sha256="0" * 64) + with pytest.raises(PersistenceIntegrityError, match="replayed decision bindings"): + await engine.store.save_approval( + forged, + decision_trace, + expected_recommendation_sha256=result.content_sha256, + expected_trace_revision=decision_trace.revision, + ) + + with pytest.raises(PersistenceIntegrityError, match="verified reviewer"): + replace(stored, reviewer_identity="forged-reviewer") + + +@pytest.mark.asyncio +async def test_store_rejects_trace_envelope_and_nested_lock_approval_forgery() -> None: + engine = _engine() + result, candidate_a, _ = await _recommend(engine, "review-store-lock-integrity") + trace = await engine.trace(result.request_id) + assert trace is not None + forged_envelope = trace.clone() + forged_envelope.trace_id = "forged-trace-id" + forged_envelope.append("human_approval", "candidate approved", { + "candidate_id": candidate_a.candidate_id, + "reviewer": "reviewer", + "note": "", + "approval_mode": ApprovalMode.STANDARD.value, + }) + forged_decision = engine._bound_approval( + result, + candidate_a, + forged_envelope, + "reviewer", + ApprovalStatus.APPROVED, + ) + with pytest.raises(PersistenceIntegrityError, match="trace envelope"): + await engine.store.save_approval( + forged_decision, + forged_envelope, + expected_recommendation_sha256=result.content_sha256, + expected_trace_revision=trace.revision, + ) + + approval = await engine.approve(result.request_id, candidate_a.candidate_id, "reviewer") + current = await engine.trace(result.request_id) + assert current is not None + lock_trace = current.clone() + lock_trace.append("lock", "approved factor result locked", { + "candidate_id": candidate_a.candidate_id, + "reviewer": "locker", + "locked_result_is_immutable": True, + "trace_remains_appendable": True, + "result_tier": candidate_a.result_tier.value, + "approval_mode": approval.mode.value, + "accepted_assumptions": candidate_a.assumptions, + "override_reason": None, + "unresolved_warnings": candidate_a.warnings, + }) + snapshot = LockedResolutionEvidenceSnapshot.from_trace( + lock_trace, + registry_anchor_sha256=result.registry_anchor_sha256 or "", + policy_anchor_sha256=result.policy_anchor_sha256 or "", + ) + forged_nested = replace( + approval, + status=ApprovalStatus.LOCKED, + candidate_content_sha256="0" * 64, + ) + forged_lock = LockedResolution( + request_id=result.request_id, + candidate=candidate_a, + reviewer="locker", + approval=forged_nested, + evidence_snapshot=snapshot, + candidate_content_sha256=candidate_a.content_sha256, + recommendation_content_sha256=result.content_sha256, + approval_content_sha256=approval.content_sha256, + ) + with pytest.raises(PersistenceIntegrityError, match="canonical stored approval"): + await engine.store.save_locked( + forged_lock, + lock_trace, + expected_recommendation_sha256=result.content_sha256, + expected_trace_revision=current.revision, + ) + + forged_candidate = replace(candidate_a, factor_value=778.0) + candidate_forged_lock = LockedResolution( + request_id=result.request_id, + candidate=forged_candidate, + reviewer="locker", + approval=replace(approval, status=ApprovalStatus.LOCKED), + evidence_snapshot=snapshot, + candidate_content_sha256=forged_candidate.content_sha256, + recommendation_content_sha256=result.content_sha256, + approval_content_sha256=approval.content_sha256, + ) + with pytest.raises(PersistenceIntegrityError, match="recommendation candidate"): + await engine.store.save_locked( + candidate_forged_lock, + lock_trace, + expected_recommendation_sha256=result.content_sha256, + expected_trace_revision=current.revision, + ) + + +@pytest.mark.asyncio +async def test_store_rejects_general_trace_envelope_rewrite_and_reject_override() -> None: + engine = _engine() + result, candidate_a, _ = await _recommend(engine, "review-general-trace-integrity") + trace = await engine.trace(result.request_id) + assert trace is not None + forged = trace.clone() + forged.trace_id = "forged-trace-id" + with pytest.raises(PersistenceIntegrityError, match="trace envelope"): + await engine.store.save_trace(forged) + + decision_trace = trace.clone() + decision_trace.append("human_approval", "candidate rejected", { + "candidate_id": candidate_a.candidate_id, + "reviewer": "reviewer", + "note": "not selected", + }) + invalid_rejection = engine._bound_approval( + result, + candidate_a, + decision_trace, + "reviewer", + ApprovalStatus.REJECTED, + note="not selected", + mode=ApprovalMode.REFERENCE_OVERRIDE, + ) + with pytest.raises(ReviewStateConflictError, match="standard approval mode"): + await engine.store.save_approval( + invalid_rejection, + decision_trace, + expected_recommendation_sha256=result.content_sha256, + expected_trace_revision=trace.revision, + ) + + +@pytest.mark.asyncio +async def test_explicit_same_revision_concurrent_mutations_have_one_stale_loser() -> None: + engine = _engine() + result, candidate_a, candidate_b = await _recommend(engine, "review-explicit-cas") + trace = await engine.trace(result.request_id) + assert trace is not None + revision = trace.revision + outcomes = await asyncio.gather( + engine.approve( + result.request_id, + candidate_a.candidate_id, + "reviewer-a", + expected_trace_revision=revision, + ), + engine.reject( + result.request_id, + candidate_b.candidate_id, + "reviewer-b", + expected_trace_revision=revision, + ), + return_exceptions=True, + ) + assert sum(isinstance(item, ApprovalRecord) for item in outcomes) == 1 + assert sum(isinstance(item, StaleReviewRevisionError) for item in outcomes) == 1 + + +def test_admin_review_api_uses_verified_actor_and_stable_conflicts() -> None: + pytest.importorskip("fastapi") + from fastapi.testclient import TestClient + + from a1_factor_engine.api import AuthorizationContext, create_admin_app, create_app + + engine = _engine() + + actor = {"value": "verified-reviewer"} + + async def allow(_headers, _permission): + return AuthorizationContext( + actor["value"], "tenant", "project", + ("resolve:execute", "review:write", "review:lock"), + ) + + with TestClient(create_admin_app(engine=engine, authorizer=allow)) as client: + resolved = client.post("/api/v1/resolve", json={ + "request_id": "review-api", + "material_name": "review material", + "quantity": 1.0, + "quantity_unit": "kg", + "geography": "CN", + "year": 2025, + "subject_type": "raw_material", + "top_k": 2, + }) + assert resolved.status_code == 200 + candidate_id = resolved.json()["candidates"][0]["candidate_id"] + revision = engine.store.traces["review-api"].revision + + form_encoded = client.post( + "/api/v1/resolutions/review-api/decisions", + data={"candidate_id": candidate_id, "decision": "approve"}, + ) + assert form_encoded.status_code == 415 + + spoofed = client.post("/api/v1/resolutions/review-api/decisions", json={ + "candidate_id": candidate_id, + "decision": "approve", + "reviewer": "spoofed", + }) + assert spoofed.status_code == 422 + + approved = client.post("/api/v1/resolutions/review-api/decisions", json={ + "candidate_id": candidate_id, + "decision": "approve", + "expected_trace_revision": revision, + }) + assert approved.status_code == 200 + assert approved.json()["reviewer_identity"] == "verified-reviewer" + + actor["value"] = "verified-locker" + + stale = client.post("/api/v1/resolutions/review-api/locks", json={ + "candidate_id": candidate_id, + "expected_trace_revision": revision, + }) + assert stale.status_code == 409 + assert stale.json()["error"]["reason_code"] == "STALE_REVIEW_REVISION" + + locked = client.post("/api/v1/resolutions/review-api/locks", json={ + "candidate_id": candidate_id, + "expected_trace_revision": revision + 1, + }) + assert locked.status_code == 200 + assert locked.json()["reviewer_identity"] == "verified-locker" + + schema = client.get("/openapi.json").json() + decision_schema = schema["components"]["schemas"]["ReviewDecisionRequestDTO"] + assert "reviewer" not in decision_schema["properties"] + assert "reviewer_identity" not in decision_schema["properties"] + assert schema["paths"]["/api/v1/resolutions/{request_id}/decisions"]["post"][ + "requestBody" + ]["content"].keys() == {"application/json"} + + public_app = create_app(engine=_engine()) + public_paths = public_app.openapi()["paths"] + assert not any(path.endswith(("/decisions", "/locks")) for path in public_paths) + with TestClient(public_app) as client: + missing = client.post( + "/api/v1/resolutions/absent/decisions", + data={"candidate_id": "x", "decision": "approve"}, + ) + assert missing.status_code == 404