"""Diagnostic-only contract/proof replay adapter shell over one candidate attempt.""" from __future__ import annotations import hashlib import json from dataclasses import dataclass from enum import Enum, unique from generate.geometric_search_run import CandidateAttempt, GeometricSearchRun from generate.kernel_facts import SourceSpan CONTRACT_PROOF_REPLAY_POLICY_VERSION = "contract_proof_replay.v1" REPLAY_ADAPTER_POLICY_VERSION = CONTRACT_PROOF_REPLAY_POLICY_VERSION _CONTRACT_REPLAY_TARGET_ALLOWLIST: dict[str, str] = { "unary_delta_transition": "problem_frame_contracts.unary_delta", "fraction_decrease": "problem_frame_contracts.fraction_decrease", "percent_partition": "problem_frame_contracts.percent_partition", } VACUOUS_PROOF_DECLARATION: tuple[str, str] = ( "proof_obligations", "none_applicable.v1", ) @unique class ReplayDisposition(str, Enum): CONTRACT_REFUSED = "contract_refused" CONTRACT_CLOSED_BUT_PROOF_REFUSED = "contract_closed_but_proof_refused" CONTRACT_AND_PROOF_CLOSED = "contract_and_proof_closed" @unique class ReplayRefusalReason(str, Enum): INVALID_REPLAY_INPUT = "invalid_replay_input" CANDIDATE_IDENTITY_MISMATCH = "candidate_identity_mismatch" CONTRACT_REPLAY_UNAVAILABLE = "contract_replay_unavailable" PROOF_REPLAY_UNAVAILABLE = "proof_replay_unavailable" UNSUPPORTED_REPLAY_POLICY = "unsupported_replay_policy" UNSUPPORTED_SCHEMA_VERSION = "unsupported_schema_version" UNSUPPORTED_PROOF_OBLIGATION = "unsupported_proof_obligation" @dataclass(frozen=True, slots=True) class ProofReplayRef: proof_obligation_id: str proof_obligation_version: str proof_replay_id: str closed: bool reason_code: str @dataclass(frozen=True, slots=True) class ReplayAdapterInput: input_digest: str replay_policy_version: str run_id: str run_policy_version: str attempt_id: str attempt_index: int candidate_digest: str candidate_reconstruction_digest: str problem_frame_digest: str original_contract_assessment_id: str candidate_organ: str residual_ids: tuple[str, ...] gate_decision_id: str budget_id: str operator_set_id: str operator_set_version: str contract_replay_target: str proof_obligation_refs: tuple[str, ...] schema_versions: tuple[tuple[str, str], ...] @dataclass(frozen=True, slots=True) class ReplayAdapterResult: replay_result_id: str replay_policy_version: str input_digest: str run_id: str attempt_id: str candidate_digest: str contract_replay_assessment_id: str proof_obligation_refs: tuple[str, ...] proof_replay_refs: tuple[ProofReplayRef, ...] replay_disposition: ReplayDisposition reason_codes: tuple[str, ...] evidence_spans: tuple[SourceSpan, ...] explanation: str @dataclass(frozen=True, slots=True) class ReplayAdapterRefusal: replay_refusal_id: str replay_policy_version: str input_digest: str | None run_id: str | None attempt_id: str | None candidate_digest: str | None replay_disposition: ReplayRefusalReason reason_codes: tuple[str, ...] explanation: str def _canonical_digest(payload: dict[str, object]) -> str: encoded = json.dumps( payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), ).encode("utf-8") return hashlib.sha256(encoded).hexdigest() def _safe_getattr(value: object, name: str) -> object: try: return getattr(value, name, None) except Exception: return None def _nonempty_text(value: object) -> bool: return isinstance(value, str) and bool(value.strip()) def _text_or_none(value: object) -> str | None: return value if isinstance(value, str) and value else None def _valid_span(span: object) -> bool: return ( isinstance(span, SourceSpan) and isinstance(span.text, str) and type(span.start) is int and type(span.end) is int and span.start >= 0 and span.end >= span.start and (span.sentence_index is None or type(span.sentence_index) is int) ) def _valid_spans(value: object) -> bool: return isinstance(value, tuple) and all(_valid_span(span) for span in value) def _span_payload(span: SourceSpan) -> dict[str, object]: return { "text": span.text, "start": span.start, "end": span.end, "sentence_index": span.sentence_index, } def _proof_ref_payload(ref: ProofReplayRef) -> dict[str, object]: return { "proof_obligation_id": ref.proof_obligation_id, "proof_obligation_version": ref.proof_obligation_version, "proof_replay_id": ref.proof_replay_id, "closed": ref.closed, "reason_code": ref.reason_code, } def _valid_proof_ref(ref: object) -> bool: if not isinstance(ref, ProofReplayRef): return False return ( _nonempty_text(ref.proof_obligation_id) and _nonempty_text(ref.proof_obligation_version) and _nonempty_text(ref.proof_replay_id) and type(ref.closed) is bool and isinstance(ref.reason_code, str) ) def _valid_proof_refs(value: object) -> bool: return isinstance(value, tuple) and all(_valid_proof_ref(ref) for ref in value) def _valid_string_tuple(value: object) -> bool: return isinstance(value, tuple) and all(_nonempty_text(item) for item in value) def _schema_versions_payload( schema_versions: tuple[tuple[str, str], ...], ) -> list[list[str]]: return [[name, version] for name, version in schema_versions] def _valid_schema_versions(value: object) -> bool: if not isinstance(value, tuple): return False if not value: return True names: list[str] = [] for entry in value: if not isinstance(entry, tuple) or len(entry) != 2: return False name, version = entry if not _nonempty_text(name) or not _nonempty_text(version): return False names.append(name) if len(names) != len(set(names)): return False return names == sorted(names) def _vacuous_proof_declared(schema_versions: tuple[tuple[str, str], ...]) -> bool: return VACUOUS_PROOF_DECLARATION in schema_versions def _contract_replay_target(candidate_organ: str) -> str | None: return _CONTRACT_REPLAY_TARGET_ALLOWLIST.get(candidate_organ) def _input_payload( *, replay_policy_version: str, run_id: str, run_policy_version: str, attempt_id: str, attempt_index: int, candidate_digest: str, candidate_reconstruction_digest: str, problem_frame_digest: str, original_contract_assessment_id: str, candidate_organ: str, residual_ids: tuple[str, ...], gate_decision_id: str, budget_id: str, operator_set_id: str, operator_set_version: str, contract_replay_target: str, proof_obligation_refs: tuple[str, ...], schema_versions: tuple[tuple[str, str], ...], ) -> dict[str, object]: return { "replay_policy_version": replay_policy_version, "run_id": run_id, "run_policy_version": run_policy_version, "attempt_id": attempt_id, "attempt_index": attempt_index, "candidate_digest": candidate_digest, "candidate_reconstruction_digest": candidate_reconstruction_digest, "problem_frame_digest": problem_frame_digest, "original_contract_assessment_id": original_contract_assessment_id, "candidate_organ": candidate_organ, "residual_ids": list(residual_ids), "gate_decision_id": gate_decision_id, "budget_id": budget_id, "operator_set_id": operator_set_id, "operator_set_version": operator_set_version, "contract_replay_target": contract_replay_target, "proof_obligation_refs": list(proof_obligation_refs), "schema_versions": _schema_versions_payload(schema_versions), } def _refusal( *, replay_policy_version: str = CONTRACT_PROOF_REPLAY_POLICY_VERSION, input_digest: str | None, run_id: str | None, attempt_id: str | None, candidate_digest: str | None, replay_disposition: ReplayRefusalReason, reason_codes: tuple[str, ...], ) -> ReplayAdapterRefusal: replay_refusal_id = _canonical_digest( { "replay_refusal_id": "", "replay_policy_version": replay_policy_version, "input_digest": input_digest, "run_id": run_id, "attempt_id": attempt_id, "candidate_digest": candidate_digest, "replay_disposition": replay_disposition.value, "reason_codes": list(reason_codes), } ) return ReplayAdapterRefusal( replay_refusal_id=replay_refusal_id, replay_policy_version=replay_policy_version, input_digest=input_digest, run_id=run_id, attempt_id=attempt_id, candidate_digest=candidate_digest, replay_disposition=replay_disposition, reason_codes=reason_codes, explanation="Replay adapter refused: " + replay_disposition.value + " (" + ", ".join(reason_codes) + ").", ) def build_replay_adapter_input( *, run: GeometricSearchRun, attempt: CandidateAttempt, candidate_digest: str, candidate_reconstruction_digest: str, problem_frame_digest: str, original_contract_assessment_id: str, candidate_organ: str, proof_obligation_refs: tuple[str, ...] = (), schema_versions: tuple[tuple[str, str], ...] = (), replay_policy_version: str = CONTRACT_PROOF_REPLAY_POLICY_VERSION, ) -> ReplayAdapterInput | ReplayAdapterRefusal: """Validate run/attempt identity and emit a replay input or refusal.""" safe_run_id = _text_or_none(_safe_getattr(run, "run_id")) safe_attempt_id = _text_or_none(_safe_getattr(attempt, "attempt_id")) safe_candidate_digest = _text_or_none(candidate_digest) if not isinstance(run, GeometricSearchRun): return _refusal( input_digest=None, run_id=safe_run_id, attempt_id=safe_attempt_id, candidate_digest=safe_candidate_digest, replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT, reason_codes=("invalid_run_type",), ) if not isinstance(attempt, CandidateAttempt): return _refusal( input_digest=None, run_id=run.run_id, attempt_id=safe_attempt_id, candidate_digest=safe_candidate_digest, replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT, reason_codes=("invalid_attempt_type",), ) if replay_policy_version != CONTRACT_PROOF_REPLAY_POLICY_VERSION: return _refusal( input_digest=None, run_id=run.run_id, attempt_id=attempt.attempt_id, candidate_digest=safe_candidate_digest, replay_policy_version=replay_policy_version, replay_disposition=ReplayRefusalReason.UNSUPPORTED_REPLAY_POLICY, reason_codes=("unsupported_replay_policy_version",), ) reasons: list[str] = [] if not _nonempty_text(run.run_id): reasons.append("missing_run_id") if not _nonempty_text(attempt.attempt_id): reasons.append("missing_attempt_id") if not _nonempty_text(candidate_digest): reasons.append("missing_candidate_digest") elif attempt.candidate_digest != candidate_digest: reasons.append("candidate_digest_mismatch") if not _nonempty_text(candidate_reconstruction_digest): reasons.append("missing_candidate_reconstruction_digest") if not _nonempty_text(problem_frame_digest): reasons.append("missing_problem_frame_digest") if not _nonempty_text(original_contract_assessment_id): reasons.append("missing_original_contract_assessment_id") if not _nonempty_text(candidate_organ): reasons.append("missing_candidate_organ") contract_replay_target = _contract_replay_target(candidate_organ) if contract_replay_target is None: reasons.append("unsupported_candidate_organ") if proof_obligation_refs and not _valid_string_tuple(proof_obligation_refs): reasons.append("invalid_proof_obligation_refs") if not _valid_schema_versions(schema_versions): reasons.append("unsupported_schema_version") attempts = run.candidate_attempts if not isinstance(attempts, tuple): reasons.append("invalid_candidate_attempts") matching_ids = [ record for record in attempts if isinstance(record, CandidateAttempt) and record.attempt_id == attempt.attempt_id ] if len(matching_ids) != 1: reasons.append("ambiguous_or_missing_attempt_id") if attempt not in attempts: reasons.append("attempt_not_in_run") attempt_index = _safe_getattr(attempt, "attempt_index") if type(attempt_index) is not int or attempt_index < 0: reasons.append("invalid_attempt_index") elif not reasons and attempt_index >= len(attempts): reasons.append("attempt_index_out_of_range") elif not reasons and attempts[attempt_index] is not attempt: reasons.append("attempt_index_mismatch") if not reasons: if attempt.input_digest != run.input_digest: reasons.append("attempt_input_digest_mismatch") if problem_frame_digest != run.problem_frame_digest: reasons.append("problem_frame_digest_mismatch") if original_contract_assessment_id != run.contract_assessment_id: reasons.append("original_contract_assessment_id_mismatch") identity_mismatch_codes = { "candidate_digest_mismatch", "ambiguous_or_missing_attempt_id", "attempt_not_in_run", "attempt_index_mismatch", "attempt_index_out_of_range", "attempt_input_digest_mismatch", "problem_frame_digest_mismatch", "original_contract_assessment_id_mismatch", } if "unsupported_schema_version" in reasons: refusal_disposition = ReplayRefusalReason.UNSUPPORTED_SCHEMA_VERSION elif any(code in identity_mismatch_codes for code in reasons): refusal_disposition = ReplayRefusalReason.CANDIDATE_IDENTITY_MISMATCH else: refusal_disposition = ReplayRefusalReason.INVALID_REPLAY_INPUT if reasons: return _refusal( input_digest=None, run_id=run.run_id if _nonempty_text(run.run_id) else None, attempt_id=attempt.attempt_id if _nonempty_text(attempt.attempt_id) else None, candidate_digest=candidate_digest if _nonempty_text(candidate_digest) else None, replay_disposition=refusal_disposition, reason_codes=tuple(reasons), ) assert isinstance(attempt_index, int) assert contract_replay_target is not None input_digest = _canonical_digest( _input_payload( replay_policy_version=replay_policy_version, run_id=run.run_id, run_policy_version=run.run_policy_version, attempt_id=attempt.attempt_id, attempt_index=attempt_index, candidate_digest=candidate_digest, candidate_reconstruction_digest=candidate_reconstruction_digest, problem_frame_digest=problem_frame_digest, original_contract_assessment_id=original_contract_assessment_id, candidate_organ=candidate_organ, residual_ids=run.residual_ids, gate_decision_id=run.gate_decision_id, budget_id=run.budget_id, operator_set_id=run.operator_set_id, operator_set_version=run.operator_set_version, contract_replay_target=contract_replay_target, proof_obligation_refs=proof_obligation_refs, schema_versions=schema_versions, ) ) return ReplayAdapterInput( input_digest=input_digest, replay_policy_version=replay_policy_version, run_id=run.run_id, run_policy_version=run.run_policy_version, attempt_id=attempt.attempt_id, attempt_index=attempt_index, candidate_digest=candidate_digest, candidate_reconstruction_digest=candidate_reconstruction_digest, problem_frame_digest=problem_frame_digest, original_contract_assessment_id=original_contract_assessment_id, candidate_organ=candidate_organ, residual_ids=run.residual_ids, gate_decision_id=run.gate_decision_id, budget_id=run.budget_id, operator_set_id=run.operator_set_id, operator_set_version=run.operator_set_version, contract_replay_target=contract_replay_target, proof_obligation_refs=proof_obligation_refs, schema_versions=schema_versions, ) def classify_replay_result( replay_input: ReplayAdapterInput, *, contract_replay_assessment_id: str | None, contract_closed: bool | None, proof_replay_refs: tuple[ProofReplayRef, ...] = (), evidence_spans: tuple[SourceSpan, ...] = (), reason_codes: tuple[str, ...] = (), explanation: str = "", ) -> ReplayAdapterResult | ReplayAdapterRefusal: """Classify one candidate replay disposition from injected authority seams.""" if not isinstance(replay_input, ReplayAdapterInput): return _refusal( input_digest=None, run_id=None, attempt_id=None, candidate_digest=None, replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT, reason_codes=("invalid_replay_input_type",), ) if replay_input.replay_policy_version != CONTRACT_PROOF_REPLAY_POLICY_VERSION: return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_policy_version=replay_input.replay_policy_version, replay_disposition=ReplayRefusalReason.UNSUPPORTED_REPLAY_POLICY, reason_codes=("unsupported_replay_policy_version",), ) if not _valid_spans(evidence_spans): return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT, reason_codes=("invalid_evidence_spans",), ) if not all(isinstance(code, str) for code in reason_codes): return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT, reason_codes=("invalid_reason_codes",), ) if contract_replay_assessment_id is None or contract_closed is None: return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_disposition=ReplayRefusalReason.CONTRACT_REPLAY_UNAVAILABLE, reason_codes=("contract_replay_unavailable",), ) if not _nonempty_text(contract_replay_assessment_id): return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_disposition=ReplayRefusalReason.CONTRACT_REPLAY_UNAVAILABLE, reason_codes=("invalid_contract_replay_assessment_id",), ) if proof_replay_refs and not _valid_proof_refs(proof_replay_refs): return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_disposition=ReplayRefusalReason.INVALID_REPLAY_INPUT, reason_codes=("invalid_proof_replay_refs",), ) if contract_closed is False: return _result( replay_input=replay_input, contract_replay_assessment_id=contract_replay_assessment_id, proof_replay_refs=(), replay_disposition=ReplayDisposition.CONTRACT_REFUSED, reason_codes=reason_codes, evidence_spans=evidence_spans, explanation=explanation, ) obligations = replay_input.proof_obligation_refs if not obligations: if _vacuous_proof_declared(replay_input.schema_versions): return _result( replay_input=replay_input, contract_replay_assessment_id=contract_replay_assessment_id, proof_replay_refs=(), replay_disposition=ReplayDisposition.CONTRACT_AND_PROOF_CLOSED, reason_codes=reason_codes, evidence_spans=evidence_spans, explanation=explanation, ) return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_disposition=ReplayRefusalReason.PROOF_REPLAY_UNAVAILABLE, reason_codes=("proof_replay_unavailable",), ) if not proof_replay_refs: return _result( replay_input=replay_input, contract_replay_assessment_id=contract_replay_assessment_id, proof_replay_refs=(), replay_disposition=ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED, reason_codes=reason_codes or ("missing_proof_replay_refs",), evidence_spans=evidence_spans, explanation=explanation, ) if len(proof_replay_refs) != len(obligations): return _result( replay_input=replay_input, contract_replay_assessment_id=contract_replay_assessment_id, proof_replay_refs=proof_replay_refs, replay_disposition=ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED, reason_codes=reason_codes or ("proof_replay_ref_count_mismatch",), evidence_spans=evidence_spans, explanation=explanation, ) for index, obligation_id in enumerate(obligations): if proof_replay_refs[index].proof_obligation_id != obligation_id: return _refusal( input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, replay_disposition=ReplayRefusalReason.UNSUPPORTED_PROOF_OBLIGATION, reason_codes=("unsupported_proof_obligation",), ) if any(not ref.closed for ref in proof_replay_refs): return _result( replay_input=replay_input, contract_replay_assessment_id=contract_replay_assessment_id, proof_replay_refs=proof_replay_refs, replay_disposition=ReplayDisposition.CONTRACT_CLOSED_BUT_PROOF_REFUSED, reason_codes=reason_codes or ("proof_not_closed",), evidence_spans=evidence_spans, explanation=explanation, ) return _result( replay_input=replay_input, contract_replay_assessment_id=contract_replay_assessment_id, proof_replay_refs=proof_replay_refs, replay_disposition=ReplayDisposition.CONTRACT_AND_PROOF_CLOSED, reason_codes=reason_codes, evidence_spans=evidence_spans, explanation=explanation, ) def _result( *, replay_input: ReplayAdapterInput, contract_replay_assessment_id: str, proof_replay_refs: tuple[ProofReplayRef, ...], replay_disposition: ReplayDisposition, reason_codes: tuple[str, ...], evidence_spans: tuple[SourceSpan, ...], explanation: str, ) -> ReplayAdapterResult: result_payload = { "replay_result_id": "", "replay_policy_version": replay_input.replay_policy_version, "input_digest": replay_input.input_digest, "run_id": replay_input.run_id, "attempt_id": replay_input.attempt_id, "candidate_digest": replay_input.candidate_digest, "contract_replay_assessment_id": contract_replay_assessment_id, "proof_obligation_refs": list(replay_input.proof_obligation_refs), "proof_replay_refs": [_proof_ref_payload(ref) for ref in proof_replay_refs], "replay_disposition": replay_disposition.value, "reason_codes": list(reason_codes), "evidence_spans": [_span_payload(span) for span in evidence_spans], } replay_result_id = _canonical_digest(result_payload) safe_explanation = explanation or ( "Replay adapter classified candidate as " + replay_disposition.value + "." ) return ReplayAdapterResult( replay_result_id=replay_result_id, replay_policy_version=replay_input.replay_policy_version, input_digest=replay_input.input_digest, run_id=replay_input.run_id, attempt_id=replay_input.attempt_id, candidate_digest=replay_input.candidate_digest, contract_replay_assessment_id=contract_replay_assessment_id, proof_obligation_refs=replay_input.proof_obligation_refs, proof_replay_refs=proof_replay_refs, replay_disposition=replay_disposition, reason_codes=reason_codes, evidence_spans=evidence_spans, explanation=safe_explanation, ) __all__ = [ "CONTRACT_PROOF_REPLAY_POLICY_VERSION", "REPLAY_ADAPTER_POLICY_VERSION", "VACUOUS_PROOF_DECLARATION", "ReplayDisposition", "ReplayRefusalReason", "ProofReplayRef", "ReplayAdapterInput", "ReplayAdapterResult", "ReplayAdapterRefusal", "build_replay_adapter_input", "classify_replay_result", ]